import os import json from datetime import datetime, time, date from fastapi import FastAPI, BackgroundTasks, HTTPException from fastapi.middleware.cors import CORSMiddleware from zoneinfo import ZoneInfo from data_updater import update_daily_data, is_trading_day from forecaster_engine import generate_predictions from signal_generator import generate_signals from t5_engine import run_t5_pipeline from forecaster_cli import run_daemon, get_prediction IST = ZoneInfo("Asia/Kolkata") MARKET_CLOSE_BUFFER = time(15, 45) # Update runs after 3:45 PM SIGNAL_TIME = time(9, 30) # Signal generation at 9:30 AM PREDICTIONS_FILE = os.path.join(os.path.dirname(__file__), "predictions.json") SIGNALS_FILE = os.path.join(os.path.dirname(__file__), "signals.json") T5_PREDICTIONS_FILE = os.path.join(os.path.dirname(__file__), "t5_predictions.json") app = FastAPI(title="HF NIFTY Forecaster Backend") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) def run_update_pipeline(): 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() except Exception as e: print(f"Pipeline error: {e}") def run_signal_pipeline(): """Run the 5-ticker signal generator.""" try: result = generate_signals() print(f"Signal generation result: {result.get('primary_signal', {}).get('action', 'UNKNOWN')}") except Exception as e: print(f"Signal pipeline error: {e}") # ── Existing Endpoints ─────────────────────────────────────────────────────── @app.get("/predictions") def get_predictions(): if not os.path.exists(PREDICTIONS_FILE): raise HTTPException(status_code=404, detail="Predictions not yet generated") with open(PREDICTIONS_FILE, "r") as f: data = json.load(f) return data @app.post("/cron/update") def cron_trigger(background_tasks: BackgroundTasks): now = datetime.now(IST) today = now.date() current_time = now.time() # 1. Check if it's a trading day 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."} # ── NEW: T5 Forecaster Endpoints ───────────────────────────────────────────── import math from fastapi.responses import JSONResponse def _sanitize_for_json(obj): """Recursively replace NaN/Inf floats with None for JSON compliance.""" if isinstance(obj, dict): return {k: _sanitize_for_json(v) for k, v in obj.items()} elif isinstance(obj, list): return [_sanitize_for_json(v) for v in obj] elif isinstance(obj, float) and (math.isnan(obj) or math.isinf(obj)): return None return obj @app.get("/t5/predictions") def get_t5_predictions(): """Get the latest first 5-minute (T5) predictions for all stocks.""" if not os.path.exists(T5_PREDICTIONS_FILE): raise HTTPException(status_code=404, detail="T5 predictions not yet generated") with open(T5_PREDICTIONS_FILE, "r") as f: data = json.load(f) # Sanitize NaN/Inf values that break FastAPI's JSON serializer data = _sanitize_for_json(data) return data @app.post("/cron/t5_update") def t5_update_trigger(background_tasks: BackgroundTasks): """ Trigger T5 prediction generation. Should be called at or after 09:20 AM IST. """ now = datetime.now(IST) today = now.date() current_time = now.time() # 1. Check if it's a trading day if not is_trading_day(today): return {"status": "skipped", "reason": f"{today} is a holiday or weekend"} # 2. Check if it's past 09:20 AM T5_UPDATE_TIME = time(9, 20) if current_time < T5_UPDATE_TIME: return {"status": "skipped", "reason": "Market first 5 minutes not completed yet. Runs after 09:20 AM IST."} background_tasks.add_task(run_t5_pipeline) return {"status": "triggered", "message": "T5 update pipeline started in the background."} @app.post("/t5/generate-now") def force_t5_generation(background_tasks: BackgroundTasks): """Force T5 prediction generation immediately, bypassing time checks.""" background_tasks.add_task(run_t5_pipeline) return { "status": "triggered", "message": "T5 generation forced. Check /t5/predictions for results.", "trigger_time": datetime.now(IST).isoformat(), } # ── NEW: NIFTY 50 Multi-Tier Forecaster Endpoints ───────────────────────────── @app.get("/nifty50") def get_nifty50_predictions(): """Get the latest high-conviction BUY predictions for NIFTY 50.""" nifty_file = os.path.join(os.path.dirname(__file__), "nifty50_predictions.json") if not os.path.exists(nifty_file): return { "last_updated": None, "total_analyzed": 0, "high_conviction_buys": 0, "predictions": [] } with open(nifty_file, "r") as f: data = json.load(f) # Filter for high conviction trades (BUY) high_conviction = [p for p in data.get("predictions", []) if p.get("Decision") == "BUY"] return { "last_updated": data.get("last_updated"), "total_analyzed": len(data.get("predictions", [])), "high_conviction_buys": len(high_conviction), "predictions": high_conviction } @app.post("/cron/nifty50_update") def nifty50_update_trigger(background_tasks: BackgroundTasks): """ Trigger the multi-tier Random Forest NIFTY 50 forecasting daemon. Should be called every two weeks. """ background_tasks.add_task(run_daemon) return {"status": "triggered", "message": "NIFTY 50 forecasting daemon started in the background."} @app.get("/predict/{ticker}") def predict_single_ticker(ticker: str): """ On-demand prediction for a single ticker. """ res = get_prediction(ticker.upper()) if "error" in res: raise HTTPException(status_code=400, detail=res["error"]) return res # ── NEW: Signal Generator Endpoints ────────────────────────────────────────── @app.get("/signals") def get_signals(): """Get the latest generated trading signals for the 5-ticker system.""" if not os.path.exists(SIGNALS_FILE): raise HTTPException(status_code=404, detail="Signals not yet generated. Trigger /cron/signal first.") with open(SIGNALS_FILE, "r") as f: data = json.load(f) return data @app.post("/cron/signal") def signal_trigger(background_tasks: BackgroundTasks): """ Trigger signal generation at 9:30 AM IST. Trains models, fetches live candles, generates BUY/SELL signals. """ now = datetime.now(IST) today = now.date() # Check if it's a trading day if not is_trading_day(today): return {"status": "skipped", "reason": f"{today} is a holiday or weekend"} # Run signal generation in background background_tasks.add_task(run_signal_pipeline) return { "status": "triggered", "message": "Signal generation pipeline started. Check /signals for results.", "trigger_time": now.isoformat(), } @app.post("/signals/generate-now") def force_signal_generation(background_tasks: BackgroundTasks): """Force signal generation immediately, bypassing time checks.""" background_tasks.add_task(run_signal_pipeline) return { "status": "triggered", "message": "Signal generation forced. Check /signals for results.", "trigger_time": datetime.now(IST).isoformat(), } @app.get("/portfolio") def get_portfolio(): """Get current portfolio status from trade journal.""" trade_log = os.path.join(os.path.dirname(__file__), "data", "live_trades.json") if not os.path.exists(trade_log): return { "starting_capital": 3692.0, "current_capital": 3692.0, "total_pnl": 0, "trades_count": 0, "win_rate": 0, } with open(trade_log, "r") as f: data = json.load(f) trades = data.get("trades", []) starting_cap = data.get("starting_capital", 3692.0) cap = starting_cap for t in trades: if "net_pnl" in t and t["net_pnl"] is not None: cap += t["net_pnl"] n_closed = len([t for t in trades if t.get("net_pnl") is not None]) n_wins = len([t for t in trades if (t.get("net_pnl") or 0) > 0]) return { "starting_capital": starting_cap, "current_capital": round(cap, 2), "total_pnl": round(cap - starting_cap, 2), "trades_count": n_closed, "win_rate": round(n_wins / n_closed * 100, 1) if n_closed > 0 else 0, "last_updated": data.get("last_updated"), } @app.get("/health") def health_check(): return {"status": "alive", "server_time_ist": datetime.now(IST).isoformat()} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)