""" Live Best Ideas fund signals sourced from IBKR (bars + portfolio + VRP cycle). Writes ``RenTech/data/logs/live_fund_signals.json`` for the command center UI. """ from __future__ import annotations import json from datetime import datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo import pandas as pd from RenTech.live.ibkr_market_data import fetch_daily_panel from RenTech.live.ibkr_session import connect_ib, disconnect_ib, ensure_positions_loaded from RenTech.live.platform_config import BrokerConfig, load_platform_config from RenTech.live.portfolio import fetch_portfolio_snapshot from RenTech.live.strategies.vrp_ibkr import VrpIbkrStrategy from RenTech.monitor.build_command_center_snapshot import ( MACRO_TICKERS, SLEEVE_LABELS, _actions_from_recommendations, _compare_etf_book, _ib_stock_mv, _read_json, ) from RenTech.monitor.strategy_catalog import STRATEGY_DEFINITIONS from RenTech.strategy_stack.combine_best_ideas_stack import FUND_WEIGHT_TABLE_TACTICAL_TSMOM from RenTech.strategy_stack.multi_strategy_manager import SPDR_SECTOR_TICKERS, SectorETFRotation from RenTech.strategy_stack.portfolio_risk_manager import ( BASE_WEIGHTS, BOND_TICKERS, TacticalAWConfig, TacticalAllWeatherManager, _momentum_from_close, ) NY = ZoneInfo("America/New_York") _REPO = Path(__file__).resolve().parents[2] LOGS = _REPO / "RenTech" / "data" / "logs" DEFAULT_OUT = LOGS / "live_fund_signals.json" TACTICAL_CONFIG_PATH = LOGS / "tactical_aw_variant_sweep_best.json" def _load_tactical_config() -> TacticalAWConfig: raw = _read_json(TACTICAL_CONFIG_PATH) if isinstance(raw, dict) and isinstance(raw.get("config"), dict): cfg = raw["config"] return TacticalAWConfig( sma_window=int(cfg.get("sma_window", 200)), mom_skip_days=int(cfg.get("mom_skip_days", 21)), mom_lookback_days=int(cfg.get("mom_lookback_days", 252)), mom_threshold=float(cfg.get("mom_threshold", 0.0)), weight_mode=str(cfg.get("weight_mode", "binary")), partial_frac=float(cfg.get("partial_frac", 0.5)), risk_on_min_invested=float(cfg.get("risk_on_min_invested", 0.0)), bond_baseline_mult=float(cfg.get("bond_baseline_mult", 1.0)), cash_annual_yield=float(cfg.get("cash_annual_yield", 0.04)), ) return TacticalAWConfig(bond_baseline_mult=0.7) def _sector_rankings_from_panel( etf_dict: dict[str, pd.DataFrame], targets: dict[str, float], *, top_k: int = 3, ) -> list[dict[str, Any]]: rows: list[tuple[str, float]] = [] for tkr, df in etf_dict.items(): if df.empty or "aqr_mom" not in df.columns: continue dfx = df.copy() dfx.index = pd.to_datetime(dfx.index).tz_localize(None) mom = float(dfx["aqr_mom"].iloc[-1]) if mom == mom: rows.append((tkr, mom)) rows.sort(key=lambda x: -x[1]) out: list[dict[str, Any]] = [] for i, (tkr, mom) in enumerate(rows, start=1): out.append( { "ticker": tkr, "rank": i, "aqr_mom": round(mom, 4), "held": tkr in targets, "in_top_k": i <= top_k, } ) return out def _tactical_gate_checks( macro_dict: dict[str, pd.DataFrame], tac_port: pd.DataFrame, config: TacticalAWConfig, ) -> list[dict[str, Any]]: if tac_port.empty: return [] port = tac_port.copy() port.index = pd.to_datetime(port.index).tz_localize(None) today = pd.Timestamp.now(tz=NY).normalize().tz_localize(None) d = port[port.index <= today].index.max() row = port.loc[d] sma_w = max(20, int(config.sma_window)) checks: list[dict[str, Any]] = [] for tkr, base_w in BASE_WEIGHTS.items(): if tkr not in macro_dict: continue df = macro_dict[tkr].copy() df.index = pd.to_datetime(df.index).tz_localize(None) close = float(df["close"].iloc[-1]) sma = float(df["close"].rolling(sma_w, min_periods=sma_w).mean().iloc[-1]) mom = float( _momentum_from_close( df["close"], skip=int(config.mom_skip_days), lookback=int(config.mom_lookback_days), ).iloc[-1] ) bw = float(base_w) * (float(config.bond_baseline_mult) if tkr in BOND_TICKERS else 1.0) tw = float(row.get(f"weight_{tkr}", 0)) checks.append( { "ticker": tkr, "close": round(close, 2), "sma": round(sma, 2), "momentum": round(mom, 4), "trend_ok": bool(close > sma), "mom_ok": bool(mom > float(config.mom_threshold)), "baseline_weight": round(bw, 4), "target_weight": round(tw, 4), } ) return checks def _sector_targets_from_panel(etf_dict: dict[str, pd.DataFrame], *, top_k: int = 3) -> tuple[dict[str, float], dict[str, Any], pd.DataFrame]: rot = SectorETFRotation() log_df = rot.generate_rebalance_log(etf_dict, top_k=top_k) meta: dict[str, Any] = { "status": "ok", "source": "ibkr_historical_bars", "top_k": top_k, "path": "live_sector_rebalance", } if log_df.empty: meta["status"] = "empty" return {}, meta, log_df log_df["effective_date"] = pd.to_datetime(log_df["effective_date"]).dt.normalize() today = pd.Timestamp.now(tz=NY).normalize().tz_localize(None) future = log_df[log_df["effective_date"] > today] if len(future): eff = future["effective_date"].min() mode = "upcoming" else: eff = log_df[log_df["effective_date"] <= today]["effective_date"].max() mode = "current" grp = log_df[(log_df["effective_date"] == eff) & (log_df["weight"].astype(float) > 0)] targets = {str(r["ticker"]): float(r["weight"]) for _, r in grp.iterrows()} sig_row = log_df[log_df["effective_date"] == eff].iloc[0] if len(log_df[log_df["effective_date"] == eff]) else None meta.update( { "mode": mode, "signal_date": str(sig_row["signal_date"]) if sig_row is not None else None, "effective_date": str(pd.Timestamp(eff).date()), "new_entries": grp.loc[grp["is_new_entry"] == True, "ticker"].astype(str).tolist(), # noqa: E712 "exits_from_prior": log_df.loc[ (log_df["effective_date"] == eff) & (log_df["is_exit_from_prior"] == True), # noqa: E712 "ticker", ] .astype(str) .tolist(), "tickers": list(targets.keys()), } ) return targets, meta, log_df def _tactical_targets_from_portfolio(port: pd.DataFrame) -> tuple[dict[str, float], dict[str, Any]]: if port.empty: return {}, {"status": "empty", "source": "ibkr_historical_bars"} port = port.copy() port.index = pd.to_datetime(port.index).tz_localize(None) today = pd.Timestamp.now(tz=NY).normalize().tz_localize(None) d = port[port.index <= today]["date"].max() if "date" in port.columns else port.index.max() row = port.loc[d] targets: dict[str, float] = {} for t in BASE_WEIGHTS: col = f"weight_{t}" if col in row: w = float(row[col]) if w > 1e-9: targets[t] = w cash_w = float(row.get("cash_weight", 1.0 - sum(targets.values()))) meta = { "status": "ok", "source": "ibkr_historical_bars", "as_of": str(pd.Timestamp(d).date()), "cash_weight": cash_w, "tickers": list(targets.keys()), "total_invested_weight": float(row.get("total_invested_weight", sum(targets.values()))), } return targets, meta def _portfolio_payload(portfolio: Any) -> dict[str, Any]: positions = [ { "symbol": p.symbol, "sec_type": p.sec_type, "expiry": p.expiry, "strike": p.strike, "right": p.right, "position": p.position, "avg_cost": p.avg_cost, "market_value": p.market_value, "unrealized_pnl": p.unrealized_pnl, "con_id": p.con_id, } for p in portfolio.positions ] return { "as_of": portfolio.as_of.isoformat(), "net_liquidation_usd": portfolio.net_liquidation_usd, "available_funds_usd": portfolio.available_funds_usd, "excess_liquidity_usd": portfolio.excess_liquidity_usd, "maintenance_margin_usd": portfolio.maintenance_margin_usd, "unrealized_pnl_usd": portfolio.unrealized_pnl_usd, "realized_pnl_today_usd": portfolio.realized_pnl_today_usd, "margin_utilization": portfolio.margin_utilization, "positions": positions, "positions_by_symbol": { sym: [x for x in positions if x["symbol"] == sym] for sym in sorted({p["symbol"] for p in positions}) }, } async def _spy_regime_gates(ib: Any) -> dict[str, Any]: from RenTech.strategy_stack.live_ibkr_trader import get_spy_sma200 from ib_insync import Stock spy = Stock("SPY", "SMART", "USD") await ib.qualifyContractsAsync(spy) sma, last = await get_spy_sma200(ib, spy) return { "spy_last": last, "spy_sma200": sma, "spy_gt_sma200": bool(last > sma), "source": "ibkr", } async def build_live_fund_signals( ib: Any, *, broker: BrokerConfig, fund_weights: dict[str, float] | None = None, run_vrp: bool = True, write_sector_csv: Path | None = None, write_tactical_csv: Path | None = None, ) -> dict[str, Any]: """ Connect must already be established. Fetches IB history and optional VRP recommend cycle. """ weights = dict(fund_weights or FUND_WEIGHT_TABLE_TACTICAL_TSMOM) wsum = sum(weights.values()) if wsum > 0: weights = {k: v / wsum for k, v in weights.items()} await ensure_positions_loaded(ib) portfolio = await fetch_portfolio_snapshot(ib) fund_nav = float(portfolio.net_liquidation_usd) ib_payload = _portfolio_payload(portfolio) ib_mv = _ib_stock_mv(ib_payload) symbols = sorted(set(SPDR_SECTOR_TICKERS) | set(MACRO_TICKERS) | {"SPY"}) print(f"[live_fund] Fetching IB daily bars for {len(symbols)} symbols …", flush=True) panels = await fetch_daily_panel(ib, symbols, duration_str="2 Y") sector_dict = {t: panels[t] for t in SPDR_SECTOR_TICKERS if t in panels} sector_targets, sector_meta, sector_log = _sector_targets_from_panel(sector_dict, top_k=3) sector_meta["rankings"] = _sector_rankings_from_panel(sector_dict, sector_targets, top_k=3) if write_sector_csv and not sector_log.empty: write_sector_csv.parent.mkdir(parents=True, exist_ok=True) sector_log.to_csv(write_sector_csv, index=False) macro_dict = {t: panels[t] for t in MACRO_TICKERS if t in panels} tac_cfg = _load_tactical_config() pm = TacticalAllWeatherManager(config=tac_cfg) tac_port = pm.build_portfolio(macro_dict) tactical_targets, tactical_meta = _tactical_targets_from_portfolio(tac_port) tactical_meta["config"] = tac_cfg.__dict__ tactical_meta["gate_checks"] = _tactical_gate_checks(macro_dict, tac_port, tac_cfg) if write_tactical_csv: write_tactical_csv.parent.mkdir(parents=True, exist_ok=True) alloc_rows = [] wcols = [c for c in tac_port.columns if c.startswith("weight_")] for dt in tac_port.index: if not wcols: break cash_w = float(tac_port.loc[dt, "cash_weight"]) for col in wcols: tkr = col.replace("weight_", "") alloc_rows.append( { "date": pd.Timestamp(dt).strftime("%Y-%m-%d"), "ticker": tkr, "weight": float(tac_port.loc[dt, col]), "cash_weight": cash_w, } ) if alloc_rows: pd.DataFrame(alloc_rows).to_csv(write_tactical_csv, index=False) spy_gates = await _spy_regime_gates(ib) recs: list[dict[str, Any]] = [] vrp_report: dict[str, Any] | None = None if run_vrp and weights.get("spy_theta", 0) > 0: print("[live_fund] VRP recommend cycle …", flush=True) from RenTech.live.platform_config import PlatformConfig, RiskConfig, StrategySlot vrp_slot = StrategySlot( id="vrp_core", enabled=True, module="RenTech.live.strategies.vrp_ibkr:VrpIbkrStrategy", priority=10, capital_budget_mode="nav_pct", capital_budget_nav_pct=float(weights.get("spy_theta", 0.22)), max_new_entries_per_day=2, max_open_positions=1, enabled_for_entries=True, config={ "strategy_config_path": "RenTech/strategy_stack/sleeve_risk_fractions.json", "state_path": "RenTech/data/live_state/vrp_core.json", "recommend_out": "RenTech/data/logs/ibkr_trade_recommendation.json", }, ) mini_cfg = PlatformConfig( path=_REPO / "RenTech/live/config/live_fund_vrp.json", name="live_fund_vrp", log_dir=LOGS / "live_platform", state_dir=_REPO / "RenTech/data/live_state", kill_switch_path=_REPO / "RenTech/live/config/KILL_SWITCH", broker=broker, risk=RiskConfig(recommend_only=True, halt_on_kill_switch=True), strategies=(vrp_slot,), ) # Run VRP in-process on shared IB connection via strategy directly from RenTech.live.protocols import StrategyContext from RenTech.live.risk_gate import evaluate_platform_risk, evaluate_strategy_risk, strategy_capital_budget today = datetime.now(NY).date() now_et = datetime.now(NY) platform_risk = evaluate_platform_risk(mini_cfg, portfolio, today, recommend_only=True, ib=ib) slot = vrp_slot budget = strategy_capital_budget(slot, portfolio) strat_risk = evaluate_strategy_risk( slot, portfolio, platform_risk, capital_budget_usd=budget, state_dir=mini_cfg.state_dir, today=today, now_et=now_et, entry_after_et="15:40", force_entry_now=False, ) ctx = StrategyContext( run_id="fund_signals", today=today, now_et=now_et, ib=ib, portfolio=portfolio, platform_risk=platform_risk, strategy_risk=strat_risk, recommend_only=True, force_entry_now=False, entry_after_et="15:40", entry_before_et=slot.entry_before_et, strategy_id=slot.id, strategy_config=dict(slot.config), capital_budget_usd=budget, repo_root=_REPO, log_dir=mini_cfg.log_dir, state_dir=mini_cfg.state_dir, max_new_entries_per_day=slot.max_new_entries_per_day, max_open_positions=slot.max_open_positions, enabled_for_entries=slot.enabled_for_entries, ) strat = VrpIbkrStrategy() report = await strat.run_cycle(ctx) vrp_report = { "ok": report.ok, "action": report.action, "messages": report.messages, "artifacts": {k: str(v) for k, v in (report.artifacts or {}).items()}, } for p in ( Path("RenTech/data/logs/ibkr_trade_recommendation.json"), Path("RenTech/data/logs/ibkr_trade_recommendation_r2a_diagonal.json"), Path("RenTech/data/logs/ibkr_trade_recommendation_r2b_spread.json"), ): data = _read_json(p) if isinstance(data, dict): recs.append({**data, "_source": p.name}) actions: list[dict[str, Any]] = [] actions.extend(_actions_from_recommendations(recs)) if weights.get("sector_momentum", 0) > 0: actions.extend( _compare_etf_book( book="sector_momentum", targets=sector_targets, ib_mv=ib_mv, fund_nav=fund_nav, fund_weight=weights["sector_momentum"], meta=sector_meta, ) ) if weights.get("tactical_aw", 0) > 0: actions.extend( _compare_etf_book( book="tactical_aw", targets=tactical_targets, ib_mv=ib_mv, fund_nav=fund_nav, fund_weight=weights["tactical_aw"], meta=tactical_meta, ) ) for key in ("vxx_regime", "macro_aw", "vxx_long_call", "tsmom"): if weights.get(key, 0) <= 0: continue label = SLEEVE_LABELS.get(key, key) defn = STRATEGY_DEFINITIONS.get(key, {}) actions.append( { "priority": 45, "action": "WATCH", "book": key, "symbol": "", "summary": f"{label}: use backtest model until IB live plugin added", "detail": { "fund_weight": weights[key], "runner": defn.get("runner", ""), "note": "Theta/VXX stack sleeves not yet driven from IB bars in this runner", }, } ) actions.sort(key=lambda a: (a["priority"], a["action"], a.get("book", ""))) return { "generated_at": datetime.now(NY).isoformat(), "data_source": "ibkr_live", "fund_nav_usd": fund_nav, "fund_weights": weights, "ibkr": {"connected": True, "portfolio": ib_payload}, "signals": { "spy_gates": spy_gates, "sector_momentum": {"targets": sector_targets, **sector_meta}, "tactical_aw": {"targets": tactical_targets, **tactical_meta, "config": tac_cfg.__dict__}, "vrp_cycle": vrp_report, "recommendations": recs, }, "actions": actions, "bars_fetched": sorted(panels.keys()), "bars_missing": sorted(set(symbols) - set(panels.keys())), } async def run_live_fund_signals_async( *, config_path: Path | None = None, run_vrp: bool = True, ) -> dict[str, Any]: cfg = load_platform_config(config_path or (_REPO / "RenTech/live/config/live_platform_default.json")) ib = await connect_ib(cfg.broker) try: payload = await build_live_fund_signals( ib, broker=cfg.broker, run_vrp=run_vrp, write_sector_csv=LOGS / "sector_momentum_live_rebalances.csv", write_tactical_csv=LOGS / "tactical_aw_live_allocations.csv", ) return payload finally: await disconnect_ib(ib) def write_live_fund_signals(path: Path, payload: dict[str, Any]) -> Path: path = path.expanduser().resolve() path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, default=str) + "\n", encoding="utf-8") return path