# -*- coding: utf-8 -*- """ Strategy Live Signal API =================================== Endpoint for the frontend to retrieve live, execution-ready Strategy Momentum signals. - Uses 175/21-day Momentum - Applies Soft-Sector Z-Score Filtering - Returns the final Top 15 correlated-sized portfolio. """ import numpy as np, pandas as pd, yfinance as yf import warnings; warnings.filterwarnings("ignore") import json # Import the shared robust universe try: from backend.alpaca_executor import UNIVERSE except ImportError: UNIVERSE = ["AAPL", "MSFT", "GOOGL", "AMZN", "META", "NVDA", "TSLA", "JPM", "V", "JNJ"] # Import sector maps try: from backtesting.strategies.v36_engine import SECTOR_MAP, SECTORS except ImportError: SECTOR_MAP = {} SECTORS = [] import os, time # Define cache file and TTL (e.g., 24 hours = 86400 seconds) CACHE_FILE = os.path.join(os.path.dirname(__file__), "strategy_cache.json") CACHE_TTL = 86400 def get_strategy_live_signals(): """Compute and return live Strategy signals.""" # Check cache first if os.path.exists(CACHE_FILE): if time.time() - os.path.getmtime(CACHE_FILE) < CACHE_TTL: try: with open(CACHE_FILE, "r") as f: return json.load(f) except Exception as e: print(f"Cache read error: {e}") pass end = pd.Timestamp.now() # Need 175 trading days (~250 calendar days), pull 300 to be safe start = end - pd.Timedelta(days=300) tickers = list(set(UNIVERSE + ["SPY"])) raw = yf.download(tickers, start=str(start.date()), end=str(end.date()), progress=False) lvl0 = raw.columns.get_level_values(0).unique().tolist() if isinstance(raw.columns, pd.MultiIndex) else [] dc = raw["Close"] if "Close" in lvl0 else raw if isinstance(dc.columns, pd.MultiIndex): dc.columns = dc.columns.get_level_values(-1) dc = dc.ffill().dropna(how="all") spy = dc["SPY"] if "SPY" in dc.columns: dc = dc.drop(columns=["SPY"]) valid_universe = [t for t in UNIVERSE if t in dc.columns and dc[t].notna().sum() > 175] # 1. Regime Check (200-day SMA) if len(spy) < 200: spy_raw = yf.download("SPY", period="1y", progress=False)["Close"] if isinstance(spy_raw, pd.DataFrame): spy_raw = spy_raw["SPY"] spy_full = spy_raw.ffill().dropna() sma200 = spy_full.rolling(200).mean() current_spy = float(spy_full.iloc[-1]) current_sma = float(sma200.iloc[-1]) else: sma200 = spy.rolling(200).mean() current_spy = float(spy.iloc[-1]) current_sma = float(sma200.iloc[-1]) is_risk_on = current_spy > current_sma # 2. Strategy Momentum (175 days skipping last 21) if len(dc) < 176: m175 = (dc.shift(min(21, len(dc)-2)) / dc.shift(len(dc)-1)) - 1 else: m175 = (dc.shift(21) / dc.shift(175)) - 1 latest_mom = m175.iloc[-1].dropna() # Calculate Consistency (63-day win rate) daily_ret = dc.pct_change() consistency = daily_ret.gt(0).where(daily_ret.notna()).rolling(63).mean() latest_cons = consistency.iloc[-1].dropna() # V68 Soft Logic: Composite Score = Momentum * Consistency valid_tks = [t for t in valid_universe if t in latest_mom.index and t in latest_cons.index] comp_scores = latest_mom[valid_tks] * latest_cons[valid_tks] # 3. Soft-Sector Neutrality (Z-Scores) z_scores = pd.Series(index=comp_scores.index, dtype=float) if SECTORS: for sector in SECTORS: sector_tks = [t for t in comp_scores.index if SECTOR_MAP.get(t) == sector] if len(sector_tks) > 1: mu = comp_scores[sector_tks].mean() sigma = comp_scores[sector_tks].std() if sigma < 1e-8: sigma = 1e-8 z_scores[sector_tks] = (comp_scores[sector_tks] - mu) / sigma elif len(sector_tks) == 1: z_scores[sector_tks[0]] = 0.0 unmapped_tks = [t for t in comp_scores.index if t not in SECTOR_MAP] if len(unmapped_tks) > 1: mu = comp_scores[unmapped_tks].mean() sigma = comp_scores[unmapped_tks].std() if sigma < 1e-8: sigma = 1e-8 z_scores[unmapped_tks] = (comp_scores[unmapped_tks] - mu) / sigma elif len(unmapped_tks) == 1: z_scores[unmapped_tks[0]] = 0.0 else: # Fallback if no sector map available: Global Z-Score mu = comp_scores.mean() sigma = comp_scores.std() z_scores = (comp_scores - mu) / sigma z_scores = z_scores.dropna().sort_values(ascending=False) top15 = z_scores.head(15) # 4. Format Output picks = [] all_universe = [] for ticker in z_scores.index: try: price = float(dc[ticker].iloc[-1]) prev = float(dc[ticker].iloc[-2]) change = ((price - prev) / prev) * 100 mom_val = float(latest_mom[ticker]) z_val = float(z_scores[ticker]) data = { "ticker": ticker, "price": round(price, 2), "change_pct": round(change, 2), "momentum_175d": round(mom_val * 100, 2), "sector_z_score": round(z_val, 2) } all_universe.append(data) if ticker in top15.index: picks.append(data) except: pass # Vol scalar estimate (using SPY as proxy) spy_rets = spy.pct_change().dropna().tail(60) rvol = float(spy_rets.std() * np.sqrt(252)) vol_scalar = min(max(0.18 / (rvol + 1e-8), 0.05), 1.0) if not is_risk_on: vol_scalar *= 0.50 result = { "engine": "Strategy", "regime": "RISK-ON" if is_risk_on else "RISK-OFF", "spy_price": round(current_spy, 2), "sma200": round(current_sma, 2), "vol_scalar": round(vol_scalar, 3), "realized_vol": round(rvol * 100, 1), "target_vol": 18.0, "picks": picks, "all_universe": all_universe, "last_updated": str(end.date()) } # Save to cache try: with open(CACHE_FILE, "w") as f: json.dump(result, f) except Exception as e: print(f"Cache write error: {e}") pass return result if __name__ == "__main__": print(json.dumps(get_strategy_live_signals(), indent=2))