# ------------------ Imports ------------------ import pandas as pd import numpy as np import pytz import json import random from datetime import datetime import gradio as gr from apscheduler.schedulers.background import BackgroundScheduler from gradio_client import Client from talib import abstract # ------------------ Configuration ------------------ CSV_FILE = "daily_indicators.csv" PREDEFINED_SYMBOLS = ["RELIANCE", "TCS", "INFY", "HDFCBANK", "ICICIBANK"] PREDEFINED_START = "2024-01-01" INTERVAL = 1440 # Daily interval CLIENT = Client("Subham9126/IME", hf_token=HF_TOKEN) MASTER_MAPPING = {'open': 'open', 'high': 'high', 'low': 'low', 'close': 'close', 'volume': 'volume'} # ------------------ Data Fetch ------------------ def fetch_data(): """Fetch historical OHLCV data from Hugging Face API for predefined symbols.""" hist_data = CLIENT.predict( ticker_input=json.dumps(PREDEFINED_SYMBOLS), start_date=PREDEFINED_START, end_date=datetime.now(pytz.timezone("Asia/Kolkata")).strftime("%Y-%m-%d"), interval=INTERVAL, batch_size=50, batch_delay=0.5, max_concurrent=50, api_name="/execute_stock_request" ) return hist_data # ------------------ JSON to DataFrame ------------------ def candles_json_to_df(json_data): """Convert Groww-like JSON to a pandas DataFrame""" records = [] ist_timezone = pytz.timezone('Asia/Kolkata') for ticker_entry in json_data.get("data", []): symbol = ticker_entry.get("ticker") candles = ticker_entry.get("data", {}).get("candles", []) for candle in candles: if isinstance(candle, list) and len(candle) >= 6: unix = candle[0] utc_datetime = datetime.utcfromtimestamp(unix) ist_datetime = utc_datetime.replace(tzinfo=pytz.utc).astimezone(ist_timezone) records.append({ "symbol": symbol, "unix": unix, "datetime": ist_datetime.strftime('%Y-%m-%d %H:%M:%S'), "open": candle[1], "high": candle[2], "low": candle[3], "close": candle[4], "volume": candle[5], }) return pd.DataFrame(records) # ------------------ TA-Lib Indicator ------------------ def talib_indicator(name, df, **kwargs): """Run any TA-Lib indicator with auto column mapping""" func = abstract.Function(name) needed_inputs = {k: MASTER_MAPPING[k] for k in func.input_names if k in MASTER_MAPPING} func.input_names = needed_inputs return func(df, **kwargs) # ------------------ Multi-symbol TA-Lib ------------------ def calculate_multi_symbol_indicators(df, indicators=None): """Calculate TA-Lib indicators for multiple symbols""" df = df.drop_duplicates(subset=['symbol', 'unix']).sort_values(['symbol','unix']).reset_index(drop=True) indicators = indicators or ['SMA', 'MACD', 'RSI', 'ADX'] processed_symbols = [] for symbol, group_df in df.groupby('symbol'): df_copy = group_df.copy().reset_index(drop=True) for col in ['open','high','low','close','volume']: df_copy[col] = df_copy[col].astype(float) ta_df = df_copy[['open','high','low','close','volume']] indicator_columns = {} for indicator in indicators: try: result = talib_indicator(indicator, ta_df) if isinstance(result, (pd.Series, np.ndarray)): indicator_columns[indicator] = result elif isinstance(result, tuple): for i, val in enumerate(result): indicator_columns[f"{indicator}_{i}"] = val except Exception as e: print(f"[WARNING] Failed {indicator} for {symbol}: {e}") continue for col_name, col_data in indicator_columns.items(): df_copy[col_name] = col_data processed_symbols.append(df_copy) return pd.concat(processed_symbols, ignore_index=True) # ------------------ Indicator Mapping ------------------ def create_indicator_mapping(columns): mapping = {} skip_cols = {'symbol','unix','datetime','open','high','low','close','volume'} indicator_cols = [c for c in columns if c not in skip_cols] from collections import defaultdict temp_map = defaultdict(list) for col in indicator_cols: if "_" in col: prefix = col.split("_")[0] temp_map[prefix].append(col) else: mapping[col] = [col] for key, values in temp_map.items(): mapping[key] = values return mapping # ------------------ Query Indicators ------------------ def query_indicators(df, tickers, indicators, date=None, start=None, end=None, mapping=None): """Return JSON-formatted indicator results""" if isinstance(tickers, str): tickers = [t.strip() for t in tickers.split(",")] if isinstance(indicators, str): indicators = [i.strip() for i in indicators.split(",")] mapping = mapping or create_indicator_mapping(df.columns.tolist()) df['datetime'] = pd.to_datetime(df['datetime']) df_filtered = df[df['symbol'].isin(tickers)] if date: df_filtered = df_filtered[df_filtered['datetime'].dt.strftime("%Y-%m-%d") == date] else: start = start or PREDEFINED_START end = end or datetime.now(pytz.timezone("Asia/Kolkata")).strftime("%Y-%m-%d") df_filtered = df_filtered[(df_filtered['datetime'].dt.strftime("%Y-%m-%d") >= start) & (df_filtered['datetime'].dt.strftime("%Y-%m-%d") <= end)] results = [] for ticker in tickers: df_ticker = df_filtered[df_filtered['symbol']==ticker] for indicator in indicators: cols = mapping.get(indicator, []) for _, row in df_ticker.iterrows(): results.append({ "ticker": ticker, "date": row['datetime'].strftime("%Y-%m-%d %H:%M:%S"), "indicator": indicator, "values": {col.replace(f"{indicator}_","") if len(cols)>1 else "value": row[col] for col in cols} }) return results # ------------------ Daily CSV Refresh ------------------ def fetch_and_store_csv(): raw_data = fetch_data() df = candles_json_to_df(raw_data) df_indicators = calculate_multi_symbol_indicators(df) df_indicators.to_csv(CSV_FILE, index=False) global indicator_mapping indicator_mapping = create_indicator_mapping(df_indicators.columns.tolist()) print(f"[INFO] CSV refreshed at {datetime.now()}") def schedule_daily_update(): ist = pytz.timezone("Asia/Kolkata") scheduler = BackgroundScheduler(timezone=ist) random_minute = random.randint(0, 30) scheduler.add_job(fetch_and_store_csv, 'cron', hour=16, minute=random_minute) scheduler.start() print(f"[INFO] Scheduled daily CSV update at 16:{random_minute:02d} IST") # ------------------ Gradio Functions ------------------ def gradio_query(tickers, indicators, date=None, start=None, end=None): df = pd.read_csv(CSV_FILE) return json.dumps(query_indicators(df, tickers, indicators, date, start, end, indicator_mapping), indent=2) def view_last_csv(): try: df = pd.read_csv(CSV_FILE) return df.tail(20).to_string() except FileNotFoundError: return "[ERROR] CSV file not found!" def manual_refresh_csv(): fetch_and_store_csv() return f"[INFO] CSV refreshed manually at {datetime.now()}" # ------------------ Initialize ------------------ fetch_and_store_csv() # Initial CSV schedule_daily_update() # Start scheduler # ------------------ Gradio UI ------------------ with gr.Blocks() as app: gr.Markdown("## TA-Lib Indicators Dashboard") with gr.Row(): tickers_input = gr.Textbox(label="Tickers (comma-separated)", value="RELIANCE,INFY") indicators_input = gr.Textbox(label="Indicators (comma-separated)", value="SMA,MACD") with gr.Row(): date_input = gr.Textbox(label="Date (optional YYYY-MM-DD)", value="") start_input = gr.Textbox(label="Start Date (optional YYYY-MM-DD)", value="") end_input = gr.Textbox(label="End Date (optional YYYY-MM-DD)", value="") output_json = gr.Code(label="Output JSON", language="json") run_button = gr.Button("Get Indicators") run_button.click( gradio_query, inputs=[tickers_input, indicators_input, date_input, start_input, end_input], outputs=output_json ) gr.Markdown("### Debug / Manual Controls") with gr.Row(): view_csv_btn = gr.Button("View Last CSV (Tail 20 rows)") csv_view_output = gr.Textbox(label="Last CSV Preview", lines=20) view_csv_btn.click(view_last_csv, outputs=csv_view_output) manual_refresh_btn = gr.Button("Manual Refresh CSV") refresh_output = gr.Textbox(label="Manual Refresh Status") manual_refresh_btn.click(manual_refresh_csv, outputs=refresh_output) app.launch()