Upload 13 files

app.py

@@ -7,9 +7,13 @@ from zoneinfo import ZoneInfo
 from data_updater import update_daily_data, is_trading_day
 from forecaster_engine import generate_predictions
 
+from t5_engine import generate_t5_predictions
+
 IST = ZoneInfo("Asia/Kolkata")
 MARKET_CLOSE_BUFFER = time(15, 45) # Update runs after 3:45 PM
+T5_RUN_BUFFER = time(9, 21) # T+5 runs after 9:21 AM
 PREDICTIONS_FILE = os.path.join(os.path.dirname(__file__), "predictions.json")
+PREDICTIONS_FILE_T5 = os.path.join(os.path.dirname(__file__), "predictions_t5.json")
 
 app = FastAPI(title="HF NIFTY Forecaster Backend")
 
@@ -20,16 +24,20 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
-def run_update_pipeline():
+def run_update_pipeline(is_t5=False):
     try:
-        # Step 1: Update data
-        res = update_daily_data()
-        if res.get("status") == "error":
-            print(f"Update failed: {res.get('reason')}")
-            return
-            
-        # Step 2: Generate predictions
-        generate_predictions()
+        if is_t5:
+            # Step 1: Generate T+5 predictions
+            generate_t5_predictions()
+        else:
+            # Step 1: Update daily data
+            res = update_daily_data()
+            if res.get("status") == "error":
+                print(f"Update failed: {res.get('reason')}")
+                return
+                
+            # Step 2: Generate T+1 predictions
+            generate_predictions()
     except Exception as e:
         print(f"Pipeline error: {e}")
 
@@ -43,6 +51,16 @@ def get_predictions():
         
     return data
 
+@app.get("/t5-predictions")
+def get_t5_predictions():
+    if not os.path.exists(PREDICTIONS_FILE_T5):
+        raise HTTPException(status_code=404, detail="T+5 Predictions not yet generated")
+    
+    with open(PREDICTIONS_FILE_T5, "r") as f:
+        data = json.load(f)
+        
+    return data
+
 @app.post("/cron/update")
 def cron_trigger(background_tasks: BackgroundTasks):
     now = datetime.now(IST)
@@ -53,14 +71,17 @@ def cron_trigger(background_tasks: BackgroundTasks):
     if not is_trading_day(today):
         return {"status": "skipped", "reason": f"{today} is a holiday or weekend"}
         
-    # 2. Check if it's past 3:45 PM
-    if current_time < MARKET_CLOSE_BUFFER:
-        return {"status": "skipped", "reason": "Market is still open or buffer not reached. Runs after 3:45 PM IST."}
-        
-    # Trigger the full pipeline in the background so Netlify doesn't timeout
-    background_tasks.add_task(run_update_pipeline)
-    
-    return {"status": "triggered", "message": "Update and forecast pipeline started in the background."}
+    # Determine which pipeline to run
+    if current_time >= MARKET_CLOSE_BUFFER:
+        # Run standard T+1 end-of-day pipeline
+        background_tasks.add_task(run_update_pipeline, is_t5=False)
+        return {"status": "triggered", "message": "End-of-day T+1 update and forecast pipeline started in the background."}
+    elif current_time >= T5_RUN_BUFFER and current_time < time(10, 0):
+        # Run T+5 pipeline at 09:21
+        background_tasks.add_task(run_update_pipeline, is_t5=True)
+        return {"status": "triggered", "message": "Morning T+5 forecast pipeline started in the background."}
+    else:
+        return {"status": "skipped", "reason": f"Current time {current_time} is not in the execution windows (09:21-10:00 for T+5, after 15:45 for T+1)."}
 
 @app.get("/health")
 def health_check():

features_t5.py

@@ -0,0 +1,167 @@
+"""
+Feature extraction for T+5 model from 1-min OHLCV DataFrames.
+Uses the 09:15-09:20 candle window to predict the day's close.
+"""
+
+import numpy as np
+import pandas as pd
+
+def _safe_fill(arr):
+    return pd.Series(arr).ffill().bfill().values
+
+def extract_semantic_features_t5(df):
+    df = df.copy()
+    df["time"] = df.index.time
+    df["date_only"] = df.index.date
+    required_times = (
+        pd.date_range("09:15", "09:20", freq="min").time.tolist()
+        + [pd.to_datetime("15:10").time()]
+    )
+    df = df[~df.index.duplicated(keep="first")]
+
+    daily_close = df.groupby("date_only")["close"].last()
+    prev_daily_close = daily_close.shift(1)
+
+    time_0921 = pd.to_datetime("09:21").time()
+    time_1200 = pd.to_datetime("12:00").time()
+    df_morning = df[(df["time"] >= time_0921) & (df["time"] <= time_1200)]
+    morning_low_per_date = df_morning.groupby("date_only")["low"].min()
+    morning_high_per_date = df_morning.groupby("date_only")["high"].max()
+
+    df_filtered = df[df["time"].isin(required_times)].copy()
+    pivot_close = df_filtered.pivot(index="date_only", columns="time", values="close")
+    pivot_open  = df_filtered.pivot(index="date_only", columns="time", values="open")
+    pivot_high  = df_filtered.pivot(index="date_only", columns="time", values="high")
+    pivot_low   = df_filtered.pivot(index="date_only", columns="time", values="low")
+    pivot_vol   = df_filtered.pivot(index="date_only", columns="time", values="volume")
+
+    time_0920 = pd.to_datetime("09:20").time()
+    time_1510 = pd.to_datetime("15:10").time()
+
+    if time_0920 not in pivot_close.columns:
+        return None, None, None
+
+    pivot_close = pivot_close.dropna(subset=[time_0920])
+    valid_dates = pivot_close.index
+    times_6m = pd.date_range("09:15", "09:20", freq="min").time
+
+    feature_dicts = []
+    metadata = {}
+
+    for date in valid_dates:
+        f = {}
+        c_series = _safe_fill(pivot_close.loc[date, times_6m].values.astype(float))
+        o_series = _safe_fill(pivot_open.loc[date, times_6m].values.astype(float))
+        h_series = _safe_fill(pivot_high.loc[date, times_6m].values.astype(float))
+        l_series = _safe_fill(pivot_low.loc[date, times_6m].values.astype(float))
+        v_series = pd.Series(pivot_vol.loc[date, times_6m].values.astype(float)).fillna(0).values
+
+        o_915 = o_series[0]
+        c_920 = c_series[-1]
+
+        pdc = prev_daily_close.get(date, np.nan)
+        f["gap"] = 0 if (pd.isna(pdc) or pdc == 0) else (o_915 / pdc) - 1.0
+        f["return_6m"] = (c_920 / o_915) - 1.0 if o_915 != 0 else 0
+        f["std_dev"] = np.std(c_series / (o_915 + 1e-8))
+
+        max_h = np.max(h_series)
+        min_l = np.min(l_series)
+        f["range_pct"] = (max_h - min_l) / (o_915 + 1e-8)
+        f["upper_shadow"] = (max_h - max(o_915, c_920)) / (o_915 + 1e-8)
+        f["lower_shadow"] = (min(o_915, c_920) - min_l) / (o_915 + 1e-8)
+        f["total_vol"] = np.sum(v_series)
+
+        vwap = np.sum(((h_series + l_series + c_series) / 3.0) * v_series) / (np.sum(v_series) + 1e-8)
+        f["vwap_dev"] = (c_920 / vwap) - 1.0 if vwap != 0 else 0
+        f["price_momentum"] = (c_series[-1] - c_series[0]) / (c_series[0] + 1e-8)
+        f["morning_trend"] = np.polyfit(np.arange(len(c_series)), c_series, 1)[0]
+
+        feature_dicts.append(f)
+
+        metadata[date] = {
+            "c_0920": c_920,
+            "h_0920": float(pivot_high.loc[date, time_0920]) if time_0920 in pivot_high.columns else c_920,
+            "l_0920": float(pivot_low.loc[date, time_0920]) if time_0920 in pivot_low.columns else c_920,
+            "v_0920": float(pivot_vol.loc[date, time_0920]) if time_0920 in pivot_vol.columns else 0,
+            "c_1510": float(pivot_close.loc[date, time_1510]) if time_1510 in pivot_close.columns else c_920,
+            "h_1510": float(pivot_high.loc[date, time_1510]) if time_1510 in pivot_high.columns else c_920,
+            "l_1510": float(pivot_low.loc[date, time_1510]) if time_1510 in pivot_low.columns else c_920,
+            "dip_low": float(morning_low_per_date.get(date, c_920)),
+            "peak_high": float(morning_high_per_date.get(date, c_920)),
+        }
+
+    X = pd.DataFrame(feature_dicts, index=valid_dates).fillna(0)
+
+    if time_1510 in pivot_close.columns:
+        target = (pivot_close[time_1510] > pivot_close[time_0920]).astype(int)
+    else:
+        target = pd.Series(0, index=valid_dates)
+
+    return X, target, metadata
+
+def extract_sequential_features_t5(df):
+    df = df.copy()
+    df["time"] = df.index.time
+    df["date_only"] = df.index.date
+    required_times = (
+        pd.date_range("09:15", "09:20", freq="min").time.tolist()
+        + [pd.to_datetime("15:10").time()]
+    )
+    df = df[~df.index.duplicated(keep="first")]
+
+    time_0921 = pd.to_datetime("09:21").time()
+    time_1200 = pd.to_datetime("12:00").time()
+    df_morning = df[(df["time"] >= time_0921) & (df["time"] <= time_1200)]
+    morning_low_per_date = df_morning.groupby("date_only")["low"].min()
+    morning_high_per_date = df_morning.groupby("date_only")["high"].max()
+
+    df_filtered = df[df["time"].isin(required_times)].copy()
+    pivot_close = df_filtered.pivot(index="date_only", columns="time", values="close")
+    pivot_high  = df_filtered.pivot(index="date_only", columns="time", values="high")
+    pivot_low   = df_filtered.pivot(index="date_only", columns="time", values="low")
+    pivot_vol   = df_filtered.pivot(index="date_only", columns="time", values="volume")
+
+    time_0920 = pd.to_datetime("09:20").time()
+    time_1510 = pd.to_datetime("15:10").time()
+
+    if time_0920 not in pivot_close.columns:
+        return None, None, None
+
+    pivot_close = pivot_close.dropna(subset=[time_0920])
+    valid_dates = pivot_close.index
+    times_6m = pd.date_range("09:15", "09:20", freq="min").time
+
+    feature_dicts = []
+    metadata = {}
+
+    for date in valid_dates:
+        f = {}
+        c_series = _safe_fill(pivot_close.loc[date, times_6m].values.astype(float))
+        v_series = pd.Series(pivot_vol.loc[date, times_6m].values.astype(float)).fillna(0).values
+        c_ref = c_series[-1]
+
+        for i, t in enumerate(times_6m):
+            f[f"ret_c_{i}"] = (c_series[i] / (c_ref + 1e-8)) - 1.0
+            f[f"raw_vol_{i}"] = v_series[i]
+        feature_dicts.append(f)
+
+        metadata[date] = {
+            "c_0920": c_ref,
+            "h_0920": float(pivot_high.loc[date, time_0920]) if time_0920 in pivot_high.columns else c_ref,
+            "l_0920": float(pivot_low.loc[date, time_0920]) if time_0920 in pivot_low.columns else c_ref,
+            "v_0920": float(pivot_vol.loc[date, time_0920]) if time_0920 in pivot_vol.columns else 0,
+            "c_1510": float(pivot_close.loc[date, time_1510]) if time_1510 in pivot_close.columns else c_ref,
+            "h_1510": float(pivot_high.loc[date, time_1510]) if time_1510 in pivot_high.columns else c_ref,
+            "l_1510": float(pivot_low.loc[date, time_1510]) if time_1510 in pivot_low.columns else c_ref,
+            "dip_low": float(morning_low_per_date.get(date, c_ref)),
+            "peak_high": float(morning_high_per_date.get(date, c_ref)),
+        }
+
+    X = pd.DataFrame(feature_dicts, index=valid_dates).fillna(0)
+
+    if time_1510 in pivot_close.columns:
+        target = (pivot_close[time_1510] > pivot_close[time_0920]).astype(int)
+    else:
+        target = pd.Series(0, index=valid_dates)
+
+    return X, target, metadata

models/ASIANPAINT_t5.joblib

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:06b6c926a9012ebbba27d892b221b4591f1f21818142acdb7bf14da832ff111a
+size 261524

models/INFY_t5.joblib

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:bea768648f602a109a2044009bc8baaf6d8879bb6d163a12722456f7894590b0
+size 75331

models/ONGC_t5.joblib

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:593f0d615caa5de3d1657048b4dc3db317741eabe27c2256651c4c503787105a
+size 139795

models/POWERGRID_t5.joblib

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:82dcdd852eb2eadffb46c8df64061027e5a338dcd3cee13033b7d69e48c653a2
+size 73524

models/TECHM_t5.joblib

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:f0e709841fbb944548a9ce70f71fad0d5b89d6ae777cbf698ab453deef22a391
+size 279604

requirements.txt

@@ -5,3 +5,5 @@ requests
 pandas_market_calendars
 pyarrow
 fastparquet
+scikit-learn
+joblib

t5_engine.py

@@ -0,0 +1,167 @@
+import os
+import json
+import time
+import requests
+import joblib
+import pandas as pd
+import numpy as np
+from datetime import datetime, date
+from zoneinfo import ZoneInfo
+from features_t5 import extract_semantic_features_t5, extract_sequential_features_t5
+
+IST = ZoneInfo("Asia/Kolkata")
+DATA_DIR = os.path.dirname(__file__)
+MODELS_DIR = os.path.join(DATA_DIR, "models")
+PREDICTIONS_FILE_T5 = os.path.join(DATA_DIR, "predictions_t5.json")
+DAILY_DATA_FILE = os.path.join(DATA_DIR, "data", "nifty50_daily.parquet")
+
+TICKERS = [
+    'ADANIENT', 'ADANIPORTS', 'APOLLOHOSP', 'ASIANPAINT', 'AXISBANK', 'BAJAJ-AUTO', 'BAJAJFINSV', 'BAJFINANCE', 
+    'BHARTIARTL', 'BPCL', 'BRITANNIA', 'CIPLA', 'COALINDIA', 'DIVISLAB', 'DRREDDY', 'EICHERMOT', 'GRASIM', 
+    'HCLTECH', 'HDFCBANK', 'HDFCLIFE', 'HEROMOTOCO', 'HINDALCO', 'HINDUNILVR', 'ICICIBANK', 'INDUSINDBK', 
+    'INFY', 'ITC', 'JSWSTEEL', 'KOTAKBANK', 'LT', 'M&M', 'MARUTI', 'NESTLEIND', 'NTPC', 'ONGC', 'POWERGRID', 
+    'RELIANCE', 'SBILIFE', 'SBIN', 'SUNPHARMA', 'TATACONSUM', 'TATAMOTORS', 'TATASTEEL', 'TCS', 'TECHM', 
+    'TITAN', 'ULTRACEMCO', 'UPL', 'WIPRO'
+]
+
+def fetch_groww_t5_data(ticker: str, start_ts: int, end_ts: int):
+    url = f"https://groww.in/v1/api/charting_service/v2/chart/exchange/NSE/segment/CASH/{ticker}?endTimeInMillis={end_ts}&intervalInMinutes=1&startTimeInMillis={start_ts}"
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
+        "Accept": "application/json"
+    }
+    
+    try:
+        response = requests.get(url, headers=headers, timeout=10)
+        if response.status_code == 200:
+            data = response.json()
+            if data and 'candles' in data and len(data['candles']) > 0:
+                rows = []
+                for c in data['candles']:
+                    rows.append({
+                        "date": datetime.fromtimestamp(c[0], IST).replace(tzinfo=None),
+                        "open": float(c[1]),
+                        "high": float(c[2]),
+                        "low": float(c[3]),
+                        "close": float(c[4]),
+                        "volume": float(c[5]),
+                        "ticker": ticker
+                    })
+                return pd.DataFrame(rows)
+        return None
+    except Exception as e:
+        print(f"Error fetching T+5 for {ticker}: {e}")
+        return None
+
+def fetch_all_t5_data(today: date):
+    # Market open 09:15 to 09:20
+    start_dt = datetime.combine(today, datetime.strptime("09:15", "%H:%M").time()).replace(tzinfo=IST)
+    end_dt = datetime.combine(today, datetime.strptime("09:25", "%H:%M").time()).replace(tzinfo=IST) # fetch slightly wider just in case
+    
+    start_ts = int(start_dt.timestamp() * 1000)
+    end_ts = int(end_dt.timestamp() * 1000)
+    
+    dfs = []
+    for ticker in TICKERS:
+        df_tick = fetch_groww_t5_data(ticker, start_ts, end_ts)
+        if df_tick is not None and not df_tick.empty:
+            dfs.append(df_tick)
+        time.sleep(0.1)
+        
+    if dfs:
+        return pd.concat(dfs, ignore_index=True)
+    return pd.DataFrame()
+
+def generate_t5_predictions():
+    now = datetime.now(IST)
+    today = now.date()
+    
+    df_live = fetch_all_t5_data(today)
+    if df_live.empty:
+        print("No live data fetched for T+5.")
+        return None
+        
+    df_live.set_index("date", inplace=True)
+    
+    # Load previous day's close for the gap feature
+    prev_closes = {}
+    if os.path.exists(DAILY_DATA_FILE):
+        df_daily = pd.read_parquet(DAILY_DATA_FILE)
+        df_daily = df_daily[df_daily['date'].dt.date < today]
+        if not df_daily.empty:
+            for ticker in TICKERS:
+                t_data = df_daily[df_daily['ticker'] == ticker]
+                if not t_data.empty:
+                    # Last row is yesterday's close
+                    t_data = t_data.sort_values("date")
+                    prev_closes[ticker] = t_data.iloc[-1]['close']
+                    
+    predictions = {}
+    
+    for ticker in TICKERS:
+        model_path = os.path.join(MODELS_DIR, f"{ticker}_t5.joblib")
+        if not os.path.exists(model_path):
+            continue
+            
+        feat_type, clf = joblib.load(model_path)
+        
+        # Build dummy dataframe for extraction
+        t_data = df_live[df_live['ticker'] == ticker].copy()
+        if t_data.empty:
+            continue
+            
+        # Insert yesterday's dummy row at 15:30 to populate prev_daily_close correctly
+        if ticker in prev_closes:
+            yday = datetime.combine(today - pd.Timedelta(days=1), datetime.strptime("15:30", "%H:%M").time())
+            t_data.loc[yday] = {"open": prev_closes[ticker], "high": prev_closes[ticker], "low": prev_closes[ticker], "close": prev_closes[ticker], "volume": 0, "ticker": ticker}
+            
+        t_data.sort_index(inplace=True)
+        
+        if feat_type == "semantic":
+            X, _, _ = extract_semantic_features_t5(t_data)
+        else:
+            X, _, _ = extract_sequential_features_t5(t_data)
+            
+        if X is None or X.empty:
+            continue
+            
+        # Get today's prediction
+        if today in X.index:
+            X_today = X.loc[[today]]
+            prob_up = clf.predict_proba(X_today)[0][1]
+            prob_dn = 1.0 - prob_up
+            
+            # Confidence threshold from tuning (0.55)
+            if prob_up > prob_dn and prob_up >= 0.55:
+                predictions[ticker] = {
+                    "prediction": "UP",
+                    "probability": round(prob_up * 100, 2),
+                    "confidence": "HIGH"
+                }
+            elif prob_dn > prob_up and prob_dn >= 0.55:
+                predictions[ticker] = {
+                    "prediction": "DOWN",
+                    "probability": round(prob_dn * 100, 2),
+                    "confidence": "HIGH"
+                }
+
+    probs = [info["probability"] for info in predictions.values()]
+    mean_accuracy = round(np.mean(probs), 2) if probs else 0.0
+    median_accuracy = round(np.median(probs), 2) if probs else 0.0
+    
+    output = {
+        "generated_at": datetime.now().isoformat(),
+        "forecast_date": today.strftime('%Y-%m-%d'),
+        "mean_accuracy": mean_accuracy,
+        "median_accuracy": median_accuracy,
+        "predictions": predictions
+    }
+    
+    with open(PREDICTIONS_FILE_T5, "w") as f:
+        json.dump(output, f, indent=4)
+        
+    print(f"Generated T+5 predictions. Saved to {PREDICTIONS_FILE_T5}")
+    return output
+
+if __name__ == "__main__":
+    generate_t5_predictions()