Spaces:
Running
Running
| 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 βββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| 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 | |
| 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 | |
| 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 | |
| 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."} | |
| 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 βββββββββββββββββββββββββββββ | |
| 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 | |
| } | |
| 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."} | |
| 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 ββββββββββββββββββββββββββββββββββββββββββ | |
| 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 | |
| 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(), | |
| } | |
| 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(), | |
| } | |
| 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"), | |
| } | |
| 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) | |