app.py

# ------------------ 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()