"""DeepQuant — deterministic rebalance engine (Phase 1: compute only, no execution). Reproduces the MarketFM Tactical target portfolio and diffs it against the live account in DOLLARS, trading only the change. No order is placed here — compute_orders returns the list the agent *would* submit. Execution (Phase 2) consumes this same output, guardrailed. """ import json import os import ssl import urllib.request import numpy as np import pandas as pd from huggingface_hub import hf_hub_download try: import certifi _CTX = ssl.create_default_context(cafile=certifi.where()) except Exception: _CTX = ssl.create_default_context() PANEL_REPO = os.environ.get("MARKETFM_REPO", "s-ttp/marketfm-panel") AV = os.environ.get("ALPHAVANTAGE_API_KEY", "") SEG_K, SECTOR_CAP, SIZE, MAX_W = 1000, 5, 50, 0.10 # top-1000 by mktcap, ≤5/sector, 50 names, 10% cap DEF_EQ, DEF_AGG, DEF_GLD = 0.60, 0.20, 0.20 # defense: 60% stocks + 20% bonds + 20% gold ETF = {"AGG": ("iShares Core US Aggregate Bond ETF", "Bonds (ETF)"), "GLD": ("SPDR Gold Shares", "Gold (ETF)")} def cap_weights(w, mw=MAX_W): w = np.asarray(w, float) for _ in range(30): over = w > mw + 1e-9 if not over.any(): break excess = (w[over] - mw).sum() w[over] = mw under = ~over if w[under].sum() <= 0: break w[under] += excess * w[under] / w[under].sum() return w DEFAULT_CFG = {"signal": ["n_roa"], "regime": "gate20", "defense": "mom60_agg20_gld20", "cadence_weeks": 4, "construction": "n60_nocap", "version": 2} CONSTRUCTIONS = { "n60_nocap": {"rule": "abs", "hard": 0.10, "buffer": 0.0, "sector_cap": 999, "size": 60}, "n50_nocap": {"rule": "abs", "hard": 0.10, "buffer": 0.0, "sector_cap": 999, "size": 50}, "n75_nocap": {"rule": "abs", "hard": 0.10, "buffer": 0.0, "sector_cap": 999, "size": 75}, "abs10_sec5": {"rule": "abs", "hard": 0.10, "buffer": 0.0, "sector_cap": 5, "size": 50}, } def cap_vec(w, caps): w = np.asarray(w, float); caps = np.asarray(caps, float) for _ in range(50): over = w > caps + 1e-9 if not over.any(): break ex = (w[over] - caps[over]).sum(); w[over] = caps[over] under = ~over room = w[under].sum() if room <= 0: break w[under] += ex * w[under] / room return w def load_config(token): """Live strategy config written by the autonomous recalibration agent (falls back to the incumbent).""" try: c = json.load(open(hf_hub_download(PANEL_REPO, "strategy_config.json", repo_type="dataset", token=token, force_download=True))) return {**DEFAULT_CFG, **{k: c[k] for k in DEFAULT_CFG if k in c}} except Exception: return dict(DEFAULT_CFG) def _spy_series(): u = f"https://www.alphavantage.co/query?function=TIME_SERIES_WEEKLY_ADJUSTED&symbol=SPY&apikey={AV}" ts = json.load(urllib.request.urlopen(u, timeout=25, context=_CTX))["Weekly Adjusted Time Series"] return pd.Series({k: float(v["5. adjusted close"]) for k, v in ts.items()}).sort_index() def _trail_dd(closes, n=52): win = list(closes[-n:]) if len(win) < 26: return 0.0 pk, m = win[0], 0.0 for p in win: pk = max(pk, p); m = max(m, 1 - p/pk) return m def detect_regime(rule="gate20"): try: s = _spy_series() p = float(s.iloc[-1]); s10 = float(s.rolling(10).mean().iloc[-1]); s20 = float(s.rolling(20).mean().iloc[-1]) s40 = float(s.rolling(40).mean().iloc[-1]); p_prev = float(s.iloc[-2]); s40_prev = float(s.rolling(40).mean().iloc[-2]) if rule in ("gate20", "gate15"): thr = 0.20 if rule == "gate20" else 0.15 off = not ((s10 < s40) and (_trail_dd(list(s.values)) >= thr)) elif rule == "single40": off = p >= s40 elif rule == "dual20/40": off = True if (p >= s20 and p >= s40) else (False if (p < s20 and p < s40) else (s10 >= s40)) elif rule == "hyst5": off = True if p > 1.05 * s40 else (False if p < 0.95 * s40 else (s10 >= s40)) elif rule == "confirm2": cur, prev = p >= s40, p_prev >= s40_prev off = cur if cur == prev else (s10 >= s40) else: # cross10/40 (incumbent) off = s10 >= s40 return {"offense": bool(off), "rule": rule, "sma10": s10, "sma40": s40, "spy": p} except Exception as e: return {"offense": True, "rule": rule, "sma10": None, "sma40": None, "spy": None, "warn": f"SPY fetch failed ({type(e).__name__}); defaulting to OFFENSE"} def _rank_sleeve(df, cols, sign=1, con=None): con = con or CONSTRUCTIONS["abs10_sec5"] df = df[df["marketcap"].notna() & (df["marketcap"] > 0)].copy() df = df.sort_values("marketcap", ascending=False).head(SEG_K) top500 = float(df["marketcap"].head(500).sum()) or 1.0 df["score"] = sum(sign * df[c].fillna(0) for c in cols) df = df.sort_values("score", ascending=False) picked, sec = [], {} for _, r in df.iterrows(): if sec.get(r["sector"], 0) >= con["sector_cap"]: continue sec[r["sector"]] = sec.get(r["sector"], 0) + 1 picked.append(r) if len(picked) >= con.get("size", SIZE): break mc = np.array([r["marketcap"] for r in picked], float) if con["rule"] == "rel": caps = np.maximum(np.minimum(con["hard"], mc / top500 + con["buffer"]), 0.02) else: caps = np.full(len(picked), con["hard"]) w = cap_vec(mc / mc.sum(), caps) return [{"ticker": r["ticker"], "name": r.get("name"), "sector": r["sector"], "weight": float(wi)} for r, wi in zip(picked, w)] def equity_sleeve(token, cols=None): cols = cols or DEFAULT_CFG["signal"] p = hf_hub_download(PANEL_REPO, "marketfm_snapshot.parquet", repo_type="dataset", token=token) return _rank_sleeve(pd.read_parquet(p), cols) def get_target(token): cfg = load_config(token) reg = detect_regime(cfg["regime"]) snap = pd.read_parquet(hf_hub_download(PANEL_REPO, "marketfm_snapshot.parquet", repo_type="dataset", token=token)) con = CONSTRUCTIONS.get(cfg.get("construction", "abs10_sec5"), CONSTRUCTIONS["abs10_sec5"]) sleeve = _rank_sleeve(snap, cfg["signal"], con=con) if reg["offense"]: rows = [dict(x) for x in sleeve] else: d = cfg["defense"] if d == "agg50_gld50": rows = [] eq_frac, agg_w, gld_w = 0.0, 0.5, 0.5 elif d == "lowvol": rows = _rank_sleeve(snap, ["n_vol_13w"], sign=-1, con=con) eq_frac, agg_w, gld_w = 1.0, 0.0, 0.0 elif d == "mom40_agg30_gld30": rows = [dict(x, weight=0.4 * x["weight"]) for x in sleeve]; eq_frac, agg_w, gld_w = 0.4, 0.3, 0.3 else: # mom60_agg20_gld20 (incumbent) rows = [dict(x, weight=DEF_EQ * x["weight"]) for x in sleeve]; eq_frac, agg_w, gld_w = DEF_EQ, DEF_AGG, DEF_GLD for sym, wv in [("AGG", agg_w), ("GLD", gld_w)]: if wv > 0: nm, sct = ETF[sym] rows.append({"ticker": sym, "name": nm, "sector": sct, "weight": wv}) weights = {r["ticker"]: r["weight"] for r in rows} return {"regime": reg, "weights": weights, "rows": rows, "config": cfg} def compute_orders(weights, positions, equity, band_pct=0.005, invest_frac=0.995, force_sells=False): """positions: {symbol: market_value $}. Returns order list (sells first), trading only the delta. invest_frac keeps a ~0.5% cash buffer so sell-proceeds slippage can't push cash negative (no margin). force_sells (set when cash < 0): sell deltas bypass the band so a thin ~5% overshoot spread across 50 names can actually be trimmed back — otherwise every per-name trim is under the band.""" band = band_pct * equity investable = equity * invest_frac orders = [] for sym in set(weights) | set(positions): tw = weights.get(sym, 0.0) tgt = tw * investable cur = positions.get(sym, 0.0) delta = tgt - cur cw = (cur / equity) if equity else 0.0 if tw == 0 and cur > 0: # dropped from basket → full exit (always) orders.append({"symbol": sym, "side": "SELL", "delta": -cur, "cur_w": cw, "tgt_w": 0.0, "exit": True}) elif cur == 0 and tgt > 1.0: # establish a new target position (always; band can't block it) orders.append({"symbol": sym, "side": "BUY", "delta": tgt, "cur_w": 0.0, "tgt_w": tw, "exit": False}) elif force_sells and delta < -1.0: # de-leverage: trims bypass the band orders.append({"symbol": sym, "side": "SELL", "delta": delta, "cur_w": cw, "tgt_w": tw, "exit": False}) elif abs(delta) < band: # churn guard: skip small adjustments to EXISTING positions continue else: orders.append({"symbol": sym, "side": "BUY" if delta > 0 else "SELL", "delta": delta, "cur_w": cw, "tgt_w": tw, "exit": False}) orders.sort(key=lambda o: (0 if o["side"] == "SELL" else 1, -abs(o["delta"]))) # sells first return orders