""" Signal Generator Backend for the 5-Ticker Intraday System. This module integrates the existing signal generator (core/) into the HF Space FastAPI backend. It: 1. Trains models on stored minute OHLCV parquet data. 2. Fetches live 09:15-09:30 candles from Groww API. 3. Generates BUY/SELL signals with confidence and share allocation. 4. Saves signals to signals.json for the frontend to consume. 5. Exposes /signals endpoint and /cron/signal trigger. """ import os import sys import json import math import logging import traceback from datetime import datetime, date as dt_date, time as dt_time from pathlib import Path from zoneinfo import ZoneInfo # Ensure core is importable sys.path.insert(0, str(Path(__file__).resolve().parent)) from core.config import ( TICKERS, STARTING_CAP, LEVERAGE, MIN_CONFIDENCE, SIGNAL_HOUR, SIGNAL_MINUTE, TRADE_LOG, DATA_DIR, ) from core.features import extract_semantic_features, extract_sequential_features from core.models import train_models from core.groww import fetch_groww_candles import pandas as pd import numpy as np IST = ZoneInfo("Asia/Kolkata") SIGNALS_FILE = os.path.join(os.path.dirname(__file__), "signals.json") logger = logging.getLogger("signal_generator") logger.setLevel(logging.DEBUG) if not logger.handlers: _ch = logging.StreamHandler(sys.stdout) _ch.setLevel(logging.INFO) _ch.setFormatter(logging.Formatter("%(asctime)s | %(levelname)-7s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S")) logger.addHandler(_ch) # ── Minute Data Updater ───────────────────────────────────────────────────── def update_minute_training_data(): """ Refresh minute OHLCV parquet files with the latest candle data from Groww. Fetches the last 5 days for each ticker and appends any new rows that aren't already in the parquet, so the training data stays current. """ import time as _time logger.info("Updating minute OHLCV training data from Groww...") updated_count = 0 for ticker in TICKERS: fpath = DATA_DIR / f"{ticker}_minute.parquet" try: df_live = fetch_groww_candles(ticker, days=5) if df_live is None or df_live.empty: logger.warning(f"[{ticker}] No live candles fetched, skipping update.") continue # Prepare live data for merge df_new = df_live.copy() df_new.index.name = "date" df_new = df_new.reset_index() df_new["date"] = pd.to_datetime(df_new["date"]) if fpath.exists(): df_existing = pd.read_parquet(fpath) df_existing["date"] = pd.to_datetime(df_existing["date"]) # Find the latest timestamp in existing data max_existing = df_existing["date"].max() # Only keep new rows that are after the existing max df_append = df_new[df_new["date"] > max_existing] if df_append.empty: continue df_merged = pd.concat([df_existing, df_append], ignore_index=True) df_merged.sort_values("date", inplace=True) df_merged.drop_duplicates(subset=["date"], keep="last", inplace=True) new_count = len(df_append) else: df_merged = df_new new_count = len(df_new) df_merged.to_parquet(fpath, index=False) logger.info(f"[{ticker}] Updated parquet with {new_count} new candles") updated_count += 1 except Exception as e: logger.error(f"[{ticker}] Failed to update minute data: {e}") logger.debug(traceback.format_exc()) _time.sleep(0.3) # Rate limit logger.info(f"Minute data update complete. {updated_count}/{len(TICKERS)} tickers refreshed.") return updated_count # ── Trade Journal ──────────────────────────────────────────────────────────── def load_trade_journal(): if os.path.exists(TRADE_LOG): try: with open(TRADE_LOG, "r") as f: data = json.load(f) if isinstance(data, dict): return data.get("trades", []) return data except Exception: pass return [] def save_trade_journal(trades): journal = { "starting_capital": STARTING_CAP, "leverage": LEVERAGE, "last_updated": datetime.now(IST).isoformat(), "trades": trades, } with open(TRADE_LOG, "w") as f: json.dump(journal, f, indent=2, default=str) def get_current_capital(trades): cap = STARTING_CAP for t in trades: if "net_pnl" in t and t["net_pnl"] is not None: cap += t["net_pnl"] return cap def already_traded_today(trades, today_str): return any(t.get("date") == today_str for t in trades) # ── Signal Generation ──────────────────────────────────────────────────────── def generate_signals(): """ Full signal generation pipeline: 0. Update minute OHLCV training data from Groww 1. Train models on parquet data 2. Fetch live candles from Groww 3. Extract features from 09:15-09:30 window 4. Generate predictions for all 5 tickers 5. Pick the best signal and allocate shares based on ₹3692 capital 6. Save to signals.json """ today = dt_date.today() today_str = today.isoformat() logger.info(f"Starting signal generation for {today_str}...") # Step 0: Update training data so models learn from recent market behavior try: update_minute_training_data() except Exception as e: logger.warning(f"Minute data update failed (non-fatal): {e}") # Step 1: Train models logger.info("Training models on historical minute data...") models = train_models(log_fn=logger.info) if not models: logger.error("No models trained. Cannot generate signals.") return {"status": "error", "reason": "model training failed"} # Step 2: Generate predictions for each ticker predictions = [] for ticker in TICKERS: if ticker not in models: logger.warning(f"[{ticker}] No trained model, skipping.") continue pipe_type, clf = models[ticker] df_live = fetch_groww_candles(ticker, days=5) if df_live is None or df_live.empty: logger.error(f"[{ticker}] Could not fetch live data, skipping.") continue try: if pipe_type == "semantic": X_live, _, meta = extract_semantic_features(df_live) else: X_live, _, meta = extract_sequential_features(df_live) except Exception as e: logger.error(f"[{ticker}] Feature extraction failed: {e}") logger.debug(traceback.format_exc()) continue if X_live is None or X_live.empty: logger.warning(f"[{ticker}] No features extracted from live data.") continue # Find today's row (or fall back to latest available) today_row = None today_meta = None for d in X_live.index: if d == today: today_row = X_live.loc[[d]] today_meta = meta.get(d) break if today_row is None: last_date = X_live.index[-1] logger.warning(f"[{ticker}] Today ({today}) not found. Using latest: {last_date}") today_row = X_live.iloc[[-1]] today_meta = meta.get(last_date) if today_meta is None: logger.warning(f"[{ticker}] No metadata for today.") continue try: prob_up = clf.predict_proba(today_row)[0][1] prob_down = 1.0 - prob_up except Exception as e: logger.error(f"[{ticker}] Prediction failed: {e}") logger.debug(traceback.format_exc()) continue predictions.append({ "ticker": ticker, "prob_up": prob_up, "prob_down": prob_down, "c_0930": today_meta["c_0930"], "h_0930": today_meta.get("h_0930", today_meta["c_0930"]), "l_0930": today_meta.get("l_0930", today_meta["c_0930"]), "v_0930": today_meta.get("v_0930", 0), }) if not predictions: logger.error("No predictions generated for any ticker.") _save_no_signal(today_str) return {"status": "error", "reason": "no predictions generated"} # Step 3: Pick best signal best = None best_conf = 0.0 best_short = False for p in predictions: if p["prob_up"] > best_conf: best_conf = p["prob_up"] best = p best_short = False if p["prob_down"] > best_conf: best_conf = p["prob_down"] best = p best_short = True if best is None or best_conf <= MIN_CONFIDENCE: logger.info("No signal above minimum confidence. Sitting in cash.") _save_no_signal(today_str) return {"status": "no_trade", "reason": "below confidence threshold"} # Step 4: Position sizing trades = load_trade_journal() capital = get_current_capital(trades) buying_power = capital * LEVERAGE entry_price = best["c_0930"] max_shares_cap = math.floor(buying_power / entry_price) if entry_price > 0 else 0 candle_vol = best["v_0930"] max_shares_liq = math.floor(candle_vol * 0.10) if candle_vol > 0 else max_shares_cap shares = min(max_shares_cap, max_shares_liq) if shares <= 0: logger.warning(f"Position size is 0 (capital={capital:.2f}, price={entry_price:.2f}).") _save_no_signal(today_str) return {"status": "no_trade", "reason": "position size is 0"} direction = "SELL" if best_short else "BUY" # Step 5: Build signal output signal = { "date": today_str, "ticker": best["ticker"], "action": direction, "confidence": round(best_conf * 100, 2), "entry_price": round(entry_price, 2), "shares": shares, "position_value": round(entry_price * shares, 2), "capital": round(capital, 2), "buying_power": round(buying_power, 2), "liquidity_capped": shares < max_shares_cap, "signal_time": datetime.now(IST).isoformat(), "status": "OPEN", } # All ticker probabilities all_tickers = [] for p in predictions: conf = max(p["prob_up"], p["prob_down"]) act = "SELL" if p["prob_down"] > p["prob_up"] else "BUY" ticker_capital_share = capital / len(predictions) ticker_bp = ticker_capital_share * LEVERAGE ticker_shares = math.floor(ticker_bp / p["c_0930"]) if p["c_0930"] > 0 else 0 ticker_vol = p["v_0930"] ticker_liq = math.floor(ticker_vol * 0.10) if ticker_vol > 0 else ticker_shares ticker_shares = min(ticker_shares, ticker_liq) all_tickers.append({ "ticker": p["ticker"], "action": act, "confidence": round(conf * 100, 2), "prob_up": round(p["prob_up"] * 100, 2), "prob_down": round(p["prob_down"] * 100, 2), "price": round(p["c_0930"], 2), "shares": ticker_shares, "position_value": round(p["c_0930"] * ticker_shares, 2), }) output = { "generated_at": datetime.now(IST).isoformat(), "forecast_date": today_str, "capital": round(capital, 2), "primary_signal": signal, "all_signals": sorted(all_tickers, key=lambda x: x["confidence"], reverse=True), } with open(SIGNALS_FILE, "w") as f: json.dump(output, f, indent=4, default=str) # Also log to trade journal trade_entry = { "date": today_str, "ticker": best["ticker"], "direction": "SHORT" if best_short else "LONG", "confidence": round(best_conf, 4), "entry_price": round(entry_price, 2), "shares": shares, "capital_before": round(capital, 2), "buying_power": round(buying_power, 2), "liquidity_capped": shares < max_shares_cap, "candle_volume": int(candle_vol), "signal_time": datetime.now(IST).isoformat(), "net_pnl": None, "exit_price": None, "status": "OPEN", "all_predictions": {p["ticker"]: {"prob_up": round(p["prob_up"], 4), "prob_down": round(p["prob_down"], 4)} for p in predictions}, } trades.append(trade_entry) save_trade_journal(trades) logger.info(f"SIGNAL: {direction} {shares}x {best['ticker']} @ Rs.{entry_price:.2f} (conf={best_conf:.2%})") return output def _save_no_signal(today_str): """Save a no-trade signal.""" output = { "generated_at": datetime.now(IST).isoformat(), "forecast_date": today_str, "capital": get_current_capital(load_trade_journal()), "primary_signal": { "date": today_str, "ticker": None, "action": "HOLD", "confidence": 0, "shares": 0, "status": "NO_TRADE", "signal_time": datetime.now(IST).isoformat(), }, "all_signals": [], } with open(SIGNALS_FILE, "w") as f: json.dump(output, f, indent=4, default=str) if __name__ == "__main__": result = generate_signals() print(json.dumps(result, indent=2, default=str))