Upload signal_generator.py

signal_generator.py

@@ -25,12 +25,15 @@ 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,
+    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")
 
@@ -44,6 +47,69 @@ if not logger.handlers:
     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():
@@ -87,6 +153,7 @@ def already_traded_today(trades, today_str):
 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
@@ -99,6 +166,12 @@ def generate_signals():
     
     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)