Spaces:
Sleeping
Sleeping
File size: 6,521 Bytes
1cd56b6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 | # -*- 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))
|