"""
Live **stock-only Best Ideas** signals from IBKR + Yahoo.

Writes ``RenTech/data/logs/live_stock_only_signals.json`` for the stock command center.
"""

from __future__ import annotations

import json
import math
from datetime import datetime
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo

import numpy as np
import pandas as pd

from RenTech.live.fund_signals import (
    _load_tactical_config,
    _portfolio_payload,
    _spy_regime_gates,
    _tactical_gate_checks,
    _tactical_targets_from_portfolio,
)
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.stock_book_utils import (
    DEFAULT_DIP_MAX_HOLDINGS,
    JOHANSEN_TRIPLETS,
    MACRO_TICKERS,
    MA_INVERSE_TICKERS,
    STOCK_ONLY_WEIGHTS,
    STOCK_SLEEVE_LABELS,
    TSMOM_TICKERS,
    VOL_EDGE_TICKERS,
    compare_sleeve_targets,
    dip_open_symbols_from_ib_mv,
    dip_per_name_budget_usd,
    sleeve_budget_usd,
)
from RenTech.monitor.build_command_center_snapshot import _ib_stock_mv
from RenTech.strategy_stack.data_loader import DataLoader
from RenTech.strategy_stack.main import _compute_daily_backtest_features
from RenTech.strategy_stack.multi_strategy_manager import _high_low_for_atr, _wilder_atr
from RenTech.strategy_stack.ma_slope_cross_sectional import (
    MaSlopeCrossSectional,
    MaSlopeCrossSectionalConfig,
)
from RenTech.strategy_stack.ma_slope_inverse_sleeve import MaSlopeInverseConfig, MaSlopeInverseSleeve
from RenTech.strategy_stack.portfolio_risk_manager import TacticalAllWeatherManager
from RenTech.strategy_stack.run_tsmom_managed_futures import (
    PORT_VOL_TARGET,
    SIGNAL_LOOKBACKS,
    TARGET_ASSET_VOL,
    UNIVERSE,
    VOL_WINDOW,
)
from RenTech.strategy_stack.run_volatility_edge_etn import VolEdgeConfig, target_weights

NY = ZoneInfo("America/New_York")
_REPO = Path(__file__).resolve().parents[2]
LOGS = _REPO / "RenTech" / "data" / "logs"
DEFAULT_OUT = LOGS / "live_stock_only_signals.json"


def _today() -> pd.Timestamp:
    return pd.Timestamp.now(tz=NY).normalize().tz_localize(None)


def _panel_from_yahoo(tickers: list[str], *, period: str = "2y") -> dict[str, pd.DataFrame]:
    loader = DataLoader()
    out: dict[str, pd.DataFrame] = {}
    for t in tickers:
        raw = loader.fetch_daily(t, period=period)
        if raw is None or raw.empty:
            continue
        df = _compute_daily_backtest_features(raw)
        df.index = pd.to_datetime(df.index).tz_localize(None)
        out[t] = df.sort_index()
    return out


def _merge_panels(ib_panel: dict[str, pd.DataFrame], yahoo_panel: dict[str, pd.DataFrame]) -> dict[str, pd.DataFrame]:
    out = dict(yahoo_panel)
    for k, v in ib_panel.items():
        if v is not None and not v.empty:
            out[k] = v
    return out


def _tsmom_targets_from_prices(prices: pd.DataFrame) -> tuple[dict[str, float], dict[str, Any]]:
    tickers = [t for t in prices.columns if t in UNIVERSE]
    prices = prices[tickers].dropna(how="all").ffill()
    rets = prices.pct_change()
    rebal_dates = prices.resample("BME").last().index
    max_lb = max(lb for lb, _ in SIGNAL_LOOKBACKS)
    weights = pd.Series(0.0, index=tickers)
    meta: dict[str, Any] = {"status": "empty", "source": "live_prices"}

    for rd in reversed(list(rebal_dates)):
        if rd not in prices.index:
            prior = prices.index[prices.index <= rd]
            if len(prior) == 0:
                continue
            rd = prior[-1]
        i_now = prices.index.get_loc(rd)
        if i_now < max_lb + VOL_WINDOW:
            continue
        signal_votes = pd.Series(0.0, index=tickers)
        valid = 0
        for lb, skip in SIGNAL_LOOKBACKS:
            i_past = i_now - lb
            i_skip = i_now - skip
            if i_past < 0 or i_skip < 0:
                continue
            price_now = prices.iloc[i_skip]
            price_past = prices.iloc[i_past]
            ret_signal = (price_now / price_past.replace(0, np.nan) - 1.0).fillna(0.0)
            signal_votes += ret_signal / (ret_signal.abs().mean() + 1e-8)
            valid += 1
        if valid == 0:
            continue
        signal_votes /= valid
        direction = np.sign(signal_votes)
        vol_window_data = rets.iloc[max(0, i_now - VOL_WINDOW) : i_now]
        realized_vol = vol_window_data.std(ddof=1) * np.sqrt(252)
        realized_vol = realized_vol.replace(0, np.nan).fillna(0.20)
        asset_weight = direction * (TARGET_ASSET_VOL / realized_vol)
        gross_vol_estimate = (asset_weight.abs() * realized_vol).sum()
        if gross_vol_estimate > 0:
            port_scale = PORT_VOL_TARGET / gross_vol_estimate
            denom = asset_weight.abs().sum()
            if denom > 0:
                port_scale = min(port_scale, 3.0 / denom)
            asset_weight *= port_scale
        gross = float(asset_weight.abs().sum())
        if gross < 1e-9:
            continue
        weights = (asset_weight / gross).fillna(0.0)
        meta = {
            "status": "ok",
            "source": "live_prices",
            "rebalance_date": str(pd.Timestamp(rd).date()),
            "gross_exposure": gross,
            "signals": {t: round(float(signal_votes.get(t, 0)), 4) for t in tickers},
        }
        break

    targets = {t: float(weights[t]) for t in tickers if abs(float(weights[t])) > 0.01}
    if not targets:
        meta["status"] = "flat"
    return targets, meta


def _vol_edge_live_targets() -> tuple[dict[str, float], dict[str, Any]]:
    from RenTech.strategy_stack.spx_regime_state_space.features import _load_vix3m_series
    from RenTech.strategy_stack.vrp_backtester import load_spy_vix_from_yfinance

    cfg = VolEdgeConfig()
    panel = load_spy_vix_from_yfinance()
    if panel.empty or len(panel) < 30:
        return {}, {"status": "empty", "error": "insufficient VIX history"}

    idx = panel.index
    spy_close = panel["close"].astype(float)
    vix = panel["vix_close"].astype(float)
    vix3m = _load_vix3m_series(idx)
    i = len(idx) - 1
    spy_s = spy_close.astype(float)
    rv = float(
        spy_s.iloc[max(0, i - cfg.rv_lookback + 1) : i + 1].pct_change().std(ddof=1) * math.sqrt(252) * 100
    )
    evrp = float(vix.iloc[i]) - rv
    vix_v = float(vix.iloc[i])
    vix3m_v = float(vix3m.iloc[i]) if i < len(vix3m) else float("nan")
    t_short, t_long = target_weights(
        variant=cfg.variant,
        evrp=evrp,
        vix=vix_v,
        vix3m=vix3m_v,
    )
    targets: dict[str, float] = {}
    if t_short > 0.001:
        targets["SVIX"] = t_short
    if t_long > 0.001:
        targets["VIXY"] = t_long
    meta = {
        "status": "ok",
        "source": "yfinance_vix",
        "as_of": str(idx[i].date()) if hasattr(idx[i], "date") else str(idx[i]),
        "evrp": round(evrp, 4),
        "vix": round(vix_v, 2),
        "vix3m": round(vix3m_v, 2) if vix3m_v == vix3m_v else None,
        "contango": bool(vix_v < vix3m_v) if vix3m_v == vix3m_v else None,
        "target_w_short": t_short,
        "target_w_long": t_long,
        "variant": cfg.variant,
    }
    return targets, meta


def _ma_slope_topn_targets(*, top_n: int = 10) -> tuple[dict[str, float], dict[str, Any]]:
    from RenTech.strategy_stack.equity_universe_loaders import load_equity_panel_dict

    cfg = MaSlopeCrossSectionalConfig(
        rank_metric="dual_product",
        rebalance="monthly",
        stop_mode="atr_trail",
        atr_multiplier=2.0,
    )
    try:
        # Positional arg name is daily_period (not period). Prefer SP100 for live speed.
        equity_dict = load_equity_panel_dict(
            universe="sp100",
            daily_period="5y",
            refresh_cache=False,
        )
    except Exception as exc:
        return {}, {"status": "error", "error": str(exc)}

    eng = MaSlopeCrossSectional(config=cfg)
    rebal = eng.generate_rebalance_log(equity_dict, top_n=top_n)
    meta: dict[str, Any] = {"status": "empty", "source": "yahoo_sp500", "top_n": top_n}
    if rebal.empty:
        return {}, meta

    rebal["effective_date"] = pd.to_datetime(rebal["effective_date"]).dt.normalize()
    today = _today()
    future = rebal[rebal["effective_date"] > today]
    if len(future):
        eff = future["effective_date"].min()
        mode = "upcoming"
    else:
        eff = rebal[rebal["effective_date"] <= today]["effective_date"].max()
        mode = "current"
    grp = rebal[(rebal["effective_date"] == eff) & (rebal["weight"].astype(float) > 0)]
    targets = {str(r["ticker"]): float(r["weight"]) for _, r in grp.iterrows()}
    meta.update(
        {
            "status": "ok",
            "mode": mode,
            "effective_date": str(pd.Timestamp(eff).date()),
            "tickers": list(targets.keys()),
            "new_entries": grp.loc[grp["is_new_entry"] == True, "ticker"].astype(str).tolist(),  # noqa: E712
            "exits_from_prior": rebal.loc[
                (rebal["effective_date"] == eff) & (rebal["is_exit_from_prior"] == True),  # noqa: E712
                "ticker",
            ]
            .astype(str)
            .tolist(),
        }
    )
    return targets, meta


def _ma_inverse_targets(spy_df: pd.DataFrame, etf_dict: dict[str, pd.DataFrame]) -> tuple[dict[str, float], dict[str, Any]]:
    cfg = MaSlopeInverseConfig(
        spy_regime="bear_dual_or_sma200",
        spy_exit="spy_slope_or_sma200",
        require_inverse_momentum=True,
        tickers_preferred=("SH",),
    )
    eng = MaSlopeInverseSleeve(config=cfg)
    pos_log = eng.generate_position_log(etf_dict, spy_df)
    meta: dict[str, Any] = {"status": "empty", "source": "live_bars"}
    if pos_log.empty:
        return {}, meta
    pos_log["date"] = pd.to_datetime(pos_log["date"]).dt.normalize()
    row = pos_log.iloc[-1]
    w = float(row.get("weight", 0) or 0)
    sym = str(row.get("ticker", "SH"))
    targets = {sym: w} if w > 0 else {}
    meta.update(
        {
            "status": "ok",
            "as_of": str(row["date"].date()),
            "regime_active": bool(w > 0),
            "ticker": sym,
        }
    )
    return targets, meta


def _johansen_triplet_targets(panel: dict[str, pd.DataFrame]) -> tuple[dict[str, float], dict[str, Any]]:
    from RenTech.strategy_stack.chan_johansen_triplet import (
        half_life_lookback,
        johansen_eigenvector_at,
        linear_zscore_units,
    )

    combined: dict[str, float] = {}
    triplets_meta: list[dict[str, Any]] = []
    for legs, label in JOHANSEN_TRIPLETS:
        missing = [t for t in legs if t not in panel]
        if missing:
            triplets_meta.append({"label": label, "status": "missing", "missing": missing})
            continue
        px = pd.DataFrame({t: panel[t]["close"] for t in legs}).dropna()
        if len(px) < 260:
            triplets_meta.append({"label": label, "status": "short_history", "n": len(px)})
            continue
        log_px = np.log(px.astype(float))
        t = len(px) - 1
        ev = johansen_eigenvector_at(log_px, t, fidelity="book", min_train=252, refit_bars=63)
        if ev is None:
            triplets_meta.append({"label": label, "status": "no_eigenvector"})
            continue
        evec = pd.Series(ev, index=legs)
        yport = (px * evec).sum(axis=1)
        lb = half_life_lookback(yport)
        units = float(linear_zscore_units(yport, lb, fidelity="causal").iloc[-1])
        leg_w = (evec * units).astype(float)
        gross = float(leg_w.abs().sum())
        if gross < 1e-9:
            triplets_meta.append({"label": label, "status": "flat", "units": units})
            continue
        norm = (leg_w / gross).to_dict()
        for sym, w in norm.items():
            combined[sym] = combined.get(sym, 0.0) + w / len(JOHANSEN_TRIPLETS)
        triplets_meta.append(
            {
                "label": label,
                "status": "ok",
                "legs": list(legs),
                "units": round(units, 3),
                "leg_weights": {k: round(float(v), 4) for k, v in norm.items()},
            }
        )
    if not combined:
        return {}, {"status": "empty", "triplets": triplets_meta}
    gross = sum(abs(v) for v in combined.values())
    targets = {k: v / gross for k, v in combined.items()} if gross > 0 else {}
    return targets, {"status": "ok", "source": "yahoo_johansen", "triplets": triplets_meta}


def _scan_cm_dip_signals(
    *,
    universe: str = "sp100",
    top_n: int = 10,
    limit_atr_mult: float = 0.9,
    profit_atr_mult: float = 0.5,
    entry_date: pd.Timestamp | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """
    CrackingMarkets dip candidates for *today's* DAY limit entries.

    Signal = last completed session with ≤−3% day, SMA200 up, ATR%/close > 3%.
    Limit = signal_close − limit_atr_mult × Wilder ATR(5). Only emits when
    ``entry_date`` is the next calendar session after the signal (i.e. today).
    """
    import universe_scanner as us  # type: ignore[import-not-found]

    tickers = us.get_sp100_tickers(universe=universe) if universe == "sp100" else us.get_sp500_tickers()
    tickers = tickers[:200]
    loader = DataLoader()
    today = (entry_date or _today()).normalize()
    today_d = today.date()

    candidates: list[dict[str, Any]] = []
    for tkr in tickers:
        try:
            raw = loader.fetch_daily(tkr, period="1y")
            if raw is None or len(raw) < 210:
                continue
            df = raw.copy()
            df.columns = [str(c).lower() for c in df.columns]
            if "adj close" in df.columns and "close" not in df.columns:
                df["close"] = df["adj close"]
            df.index = pd.to_datetime(df.index).tz_localize(None)
            df = df.sort_index()
            if "close" not in df.columns or len(df) < 210:
                continue

            close_s = df["close"].astype(float)
            hi, lo = _high_low_for_atr(df, close_s)
            atr_s = _wilder_atr(hi, lo, close_s, 5)
            sma200 = close_s.rolling(200, min_periods=200).mean()

            last_dt = pd.Timestamp(df.index[-1]).normalize()
            # Prefer prior completed session as signal when today's bar is already present.
            if last_dt.date() >= today_d and len(df) >= 2:
                sig_i = -2
            else:
                sig_i = -1
            if abs(sig_i) >= len(df):
                continue
            sig_dt = pd.Timestamp(df.index[sig_i]).normalize()
            # Entry is the next business day after signal; only trade today.
            next_entry = (sig_dt + pd.offsets.BDay(1)).normalize()
            if next_entry.date() != today_d:
                continue

            sig_close = float(close_s.iloc[sig_i])
            prior_close = float(close_s.iloc[sig_i - 1])
            if prior_close <= 0 or not math.isfinite(sig_close):
                continue
            ret = sig_close / prior_close - 1.0
            atr5 = float(atr_s.iloc[sig_i])
            sma_v = float(sma200.iloc[sig_i])
            if not math.isfinite(atr5) or atr5 <= 0:
                continue
            atr_pct = atr5 / sig_close * 100.0
            if ret > -0.03:
                continue
            if not math.isfinite(sma_v) or sig_close <= sma_v:
                continue
            if atr_pct <= 3.0:
                continue
            limit_px = sig_close - float(limit_atr_mult) * atr5
            profit_px = limit_px + float(profit_atr_mult) * atr5
            candidates.append(
                {
                    "ticker": str(tkr).upper(),
                    "signal_date": str(sig_dt.date()),
                    "entry_date": str(today_d),
                    "drop_pct": round(ret * 100, 2),
                    "atr_pct": round(atr_pct, 2),
                    "atr_usd": round(atr5, 4),
                    "limit_price": round(limit_px, 2),
                    "profit_target": round(profit_px, 2),
                    "prior_close": round(sig_close, 2),
                    "limit_atr_mult": float(limit_atr_mult),
                    "profit_atr_mult": float(profit_atr_mult),
                }
            )
        except Exception:
            continue
    candidates.sort(key=lambda x: -x["atr_pct"])
    picks = candidates[:top_n]
    meta = {
        "status": "ok" if picks else "none",
        "source": "yahoo_scan_atr5",
        "universe": universe,
        "n_candidates": len(candidates),
        "entry_date": str(today_d),
        "signals": picks,
    }
    return picks, meta


def _dip_actions(
    signals: list[dict[str, Any]],
    *,
    fund_nav: float,
    fund_weight: float,
    fund_scale: float,
    max_holdings: int = DEFAULT_DIP_MAX_HOLDINGS,
    open_symbols: set[str] | None = None,
) -> list[dict[str, Any]]:
    """
    Emit DAY limit ADD tickets sized at ``sleeve / max_holdings``.

    Skips names already open and caps new adds to remaining slots.
    """
    open_syms = {str(s).upper() for s in (open_symbols or set())}
    sleeve_nav = sleeve_budget_usd(fund_nav, fund_weight, fund_scale)
    per_name = dip_per_name_budget_usd(
        fund_nav, fund_weight, fund_scale, max_holdings=max_holdings
    )
    slots = max(0, int(max_holdings) - len(open_syms))
    base_detail = {
        "sleeve_budget_usd": round(sleeve_nav, 2),
        "per_name_budget_usd": round(per_name, 2),
        "max_holdings": int(max_holdings),
        "open_count": len(open_syms),
        "open_symbols": sorted(open_syms),
        "slots_remaining": slots,
    }

    if slots <= 0:
        return [
            {
                "priority": 40,
                "action": "HOLD",
                "book": "equity_dip",
                "symbol": "",
                "summary": (
                    f"Dip sleeve full ({len(open_syms)}/{int(max_holdings)}) — skip new DAY limits"
                ),
                "detail": base_detail,
            }
        ]

    if not signals:
        return [
            {
                "priority": 55,
                "action": "WATCH",
                "book": "equity_dip",
                "symbol": "",
                "summary": "No CM dip signals for today's entry session",
                "detail": base_detail,
            }
        ]

    fresh = [s for s in signals if str(s.get("ticker", "")).upper() not in open_syms]
    take = fresh[:slots]
    if not take:
        return [
            {
                "priority": 50,
                "action": "HOLD",
                "book": "equity_dip",
                "symbol": "",
                "summary": (
                    f"All {len(signals)} dip candidates already open "
                    f"({len(open_syms)}/{int(max_holdings)})"
                ),
                "detail": {**base_detail, "n_signals": len(signals)},
            }
        ]

    actions: list[dict[str, Any]] = []
    for sig in take:
        tkr = str(sig["ticker"]).upper()
        lim = float(sig.get("limit_price") or 0.0)
        actions.append(
            {
                "priority": 15,
                "action": "ADD",
                "book": "equity_dip",
                "symbol": tkr,
                "summary": f"CM dip limit buy {tkr} @ {lim:.2f} (~${per_name:,.0f})",
                "detail": {
                    **sig,
                    "ticker": tkr,
                    "target_notional_usd": round(per_name, 2),
                    "order_type": "LMT",
                    "time_in_force": "DAY",
                    **base_detail,
                },
            }
        )
    if len(fresh) > slots:
        actions.append(
            {
                "priority": 45,
                "action": "WATCH",
                "book": "equity_dip",
                "symbol": "",
                "summary": (
                    f"Skipped {len(fresh) - slots} dip candidates — only {slots} slots left"
                ),
                "detail": {
                    **base_detail,
                    "skipped": [str(s["ticker"]).upper() for s in fresh[slots:]],
                },
            }
        )
    return actions


def _qs_actionable_4_live_targets(
    spy_df: pd.DataFrame,
) -> tuple[dict[str, float], dict[str, Any]]:
    """Equal-weight SPY exposure when any actionable-4 rule is active today."""
    from RenTech.strategy_stack.qs_systematic_library import QS_ACTIONABLE_4, run_systematic_strategy

    if spy_df is None or spy_df.empty:
        return {}, {"status": "empty", "active_rules": []}

    panels = {"spy": spy_df}
    idx = spy_df.index
    active: list[str] = []
    for sid in QS_ACTIONABLE_4:
        r = run_systematic_strategy(sid, panels, idx)
        if float(r.iloc[-1]) != 0.0:
            active.append(sid)
    if not active:
        return {}, {"status": "cash", "active_rules": [], "preset": "actionable-4"}
    w = len(active) / float(len(QS_ACTIONABLE_4))
    return {"SPY": w}, {"status": "active", "active_rules": active, "preset": "actionable-4", "spy_weight": w}


async def build_stock_only_signals(
    ib: Any | None,
    *,
    broker: BrokerConfig | None = None,
    fund_weights: dict[str, float] | None = None,
    fund_scale: float = 1.5,
    fund_nav: float | None = None,
    use_yahoo: bool = True,
    renormalize_weights: bool = True,
) -> dict[str, Any]:
    weights = dict(fund_weights or STOCK_ONLY_WEIGHTS)
    if renormalize_weights:
        wsum = sum(weights.values())
        if wsum > 0:
            weights = {k: v / wsum for k, v in weights.items()}
    # Drop zero / missing sleeves so expensive paths are skipped
    weights = {k: float(v) for k, v in weights.items() if float(v) > 0}

    ib_payload: dict[str, Any] | None = None
    ib_mv: dict[str, float] = {}
    resolved_nav = float(fund_nav) if fund_nav is not None else 100_000.0

    if ib is not None:
        await ensure_positions_loaded(ib)
        portfolio = await fetch_portfolio_snapshot(ib)
        resolved_nav = float(portfolio.net_liquidation_usd)
        ib_payload = _portfolio_payload(portfolio)
        ib_mv = _ib_stock_mv(ib_payload)

    fund_nav = resolved_nav
    ib_panel: dict[str, pd.DataFrame] = {}
    if ib is not None:
        ib_symbols = sorted(
            set(MACRO_TICKERS)
            | set(TSMOM_TICKERS)
            | set(VOL_EDGE_TICKERS)
            | set(MA_INVERSE_TICKERS)
            | {"SPY"}
        )
        print(f"[stock_only] IB daily bars for {len(ib_symbols)} symbols …", flush=True)
        ib_panel = await fetch_daily_panel(ib, ib_symbols, duration_str="2 Y")

    yahoo_panel: dict[str, pd.DataFrame] = {}
    if use_yahoo:
        johansen_syms = (
            sorted({t for legs, _ in JOHANSEN_TRIPLETS for t in legs})
            if weights.get("johansen_etf", 0) > 0
            else []
        )
        macro_yahoo = [t for t in MACRO_TICKERS if t not in ib_panel]
        yahoo_syms = sorted(
            set(johansen_syms)
            | set(macro_yahoo)
            | (set(TSMOM_TICKERS) if weights.get("tsmom", 0) > 0 else set())
            | (set(VOL_EDGE_TICKERS) if weights.get("vol_edge", 0) > 0 else set())
            | (set(MA_INVERSE_TICKERS) if weights.get("ma_slope_inverse", 0) > 0 else set())
            | ({"SPY"} if weights.get("qs_actionable_etf", 0) > 0 or weights.get("tactical_aw", 0) > 0 else set())
        )
        # Always pull macro ETFs when tactical AW is on
        if weights.get("tactical_aw", 0) > 0:
            yahoo_syms = sorted(set(yahoo_syms) | set(MACRO_TICKERS))
        print(f"[stock_only] Yahoo bars for {len(yahoo_syms)} symbols …", flush=True)
        yahoo_panel = _panel_from_yahoo(yahoo_syms, period="2y") if yahoo_syms else {}

    panels = _merge_panels(ib_panel, yahoo_panel)
    macro_dict = {t: panels[t] for t in MACRO_TICKERS if t in panels}
    spy_df = panels.get("SPY", macro_dict.get("SPY"))

    tac_cfg = _load_tactical_config()
    pm = TacticalAllWeatherManager(config=tac_cfg)
    tac_port = pm.build_portfolio(macro_dict) if macro_dict else pd.DataFrame()
    tactical_targets, tactical_meta = (
        _tactical_targets_from_portfolio(tac_port)
        if weights.get("tactical_aw", 0) > 0
        else ({}, {"status": "skipped"})
    )
    if macro_dict and weights.get("tactical_aw", 0) > 0:
        tactical_meta["gate_checks"] = _tactical_gate_checks(macro_dict, tac_port, tac_cfg)

    tsmom_prices = pd.DataFrame({t: panels[t]["close"] for t in TSMOM_TICKERS if t in panels}).dropna(how="all")
    tsmom_targets, tsmom_meta = (
        _tsmom_targets_from_prices(tsmom_prices)
        if weights.get("tsmom", 0) > 0 and not tsmom_prices.empty
        else ({}, {"status": "skipped" if weights.get("tsmom", 0) <= 0 else "empty"})
    )

    if weights.get("vol_edge", 0) > 0:
        vol_targets, vol_meta = _vol_edge_live_targets()
    else:
        vol_targets, vol_meta = {}, {"status": "skipped"}

    if weights.get("ma_slope_topn", 0) > 0:
        ma_targets, ma_meta = _ma_slope_topn_targets()
    else:
        ma_targets, ma_meta = {}, {"status": "skipped"}

    sh_dict = {t: panels[t] for t in MA_INVERSE_TICKERS if t in panels}
    inverse_targets, inverse_meta = (
        _ma_inverse_targets(spy_df, sh_dict)
        if weights.get("ma_slope_inverse", 0) > 0 and spy_df is not None and sh_dict
        else ({}, {"status": "skipped" if weights.get("ma_slope_inverse", 0) <= 0 else "empty"})
    )

    if weights.get("johansen_etf", 0) > 0:
        johansen_panel = {**yahoo_panel, **{k: v for k, v in panels.items() if k not in yahoo_panel}}
        johansen_targets, johansen_meta = _johansen_triplet_targets(johansen_panel)
    else:
        johansen_targets, johansen_meta = {}, {"status": "skipped"}

    if weights.get("equity_dip", 0) > 0:
        dip_signals, dip_meta = _scan_cm_dip_signals()
    else:
        dip_signals, dip_meta = [], {"status": "skipped"}
    dip_open = dip_open_symbols_from_ib_mv(ib_mv)
    dip_meta = {**dip_meta, "open_symbols": sorted(dip_open), "open_count": len(dip_open)}
    qs_targets, qs_meta = (
        _qs_actionable_4_live_targets(spy_df)
        if weights.get("qs_actionable_etf", 0) > 0 and spy_df is not None
        else ({}, {"status": "skipped" if weights.get("qs_actionable_etf", 0) <= 0 else "empty"})
    )
    spy_gates = await _spy_regime_gates(ib) if ib is not None else {}

    actions: list[dict[str, Any]] = []
    if weights.get("tactical_aw", 0) > 0 and tactical_targets:
        actions.extend(
            compare_sleeve_targets(
                book="tactical_aw",
                targets=tactical_targets,
                ib_mv=ib_mv,
                fund_nav=fund_nav,
                fund_weight=weights["tactical_aw"],
                fund_scale=fund_scale,
                meta=tactical_meta,
            )
        )
    if weights.get("vol_edge", 0) > 0 and vol_targets:
        actions.extend(
            compare_sleeve_targets(
                book="vol_edge",
                targets=vol_targets,
                ib_mv=ib_mv,
                fund_nav=fund_nav,
                fund_weight=weights["vol_edge"],
                fund_scale=fund_scale,
                meta=vol_meta,
            )
        )
    if weights.get("tsmom", 0) > 0 and tsmom_targets:
        actions.extend(
            compare_sleeve_targets(
                book="tsmom",
                targets=tsmom_targets,
                ib_mv=ib_mv,
                fund_nav=fund_nav,
                fund_weight=weights["tsmom"],
                fund_scale=fund_scale,
                meta=tsmom_meta,
                allow_short=True,
            )
        )
    if weights.get("ma_slope_topn", 0) > 0 and ma_targets:
        actions.extend(
            compare_sleeve_targets(
                book="ma_slope_topn",
                targets=ma_targets,
                ib_mv=ib_mv,
                fund_nav=fund_nav,
                fund_weight=weights["ma_slope_topn"],
                fund_scale=fund_scale,
                meta=ma_meta,
            )
        )
    if weights.get("ma_slope_inverse", 0) > 0:
        if inverse_targets:
            actions.extend(
                compare_sleeve_targets(
                    book="ma_slope_inverse",
                    targets=inverse_targets,
                    ib_mv=ib_mv,
                    fund_nav=fund_nav,
                    fund_weight=weights["ma_slope_inverse"],
                    fund_scale=fund_scale,
                    meta=inverse_meta,
                )
            )
        else:
            actions.append(
                {
                    "priority": 70,
                    "action": "HOLD",
                    "book": "ma_slope_inverse",
                    "symbol": "CASH",
                    "summary": "SH hedge off — SPY regime clear",
                    "detail": inverse_meta,
                }
            )
    if weights.get("johansen_etf", 0) > 0 and johansen_targets:
        actions.extend(
            compare_sleeve_targets(
                book="johansen_etf",
                targets=johansen_targets,
                ib_mv=ib_mv,
                fund_nav=fund_nav,
                fund_weight=weights["johansen_etf"],
                fund_scale=fund_scale,
                meta=johansen_meta,
                allow_short=True,
            )
        )
    if weights.get("equity_dip", 0) > 0:
        actions.extend(
            _dip_actions(
                dip_signals,
                fund_nav=fund_nav,
                fund_weight=weights["equity_dip"],
                fund_scale=fund_scale,
                max_holdings=DEFAULT_DIP_MAX_HOLDINGS,
                open_symbols=dip_open,
            )
        )
    if weights.get("qs_actionable_etf", 0) > 0 and qs_targets:
        actions.extend(
            compare_sleeve_targets(
                book="qs_actionable_etf",
                targets=qs_targets,
                ib_mv=ib_mv,
                fund_nav=fund_nav,
                fund_weight=weights["qs_actionable_etf"],
                fund_scale=fund_scale,
                meta=qs_meta,
            )
        )
    elif weights.get("qs_actionable_etf", 0) > 0:
        actions.append(
            {
                "priority": 75,
                "action": "HOLD",
                "book": "qs_actionable_etf",
                "symbol": "CASH",
                "summary": "QS actionable-4 off — no SPY calendar/overnight signal today",
                "detail": qs_meta,
            }
        )

    actions.sort(key=lambda a: (a["priority"], a["action"], a.get("book", ""), a.get("symbol", "")))

    sleeve_budgets = {
        k: round(sleeve_budget_usd(fund_nav, w, fund_scale), 2)
        for k, w in weights.items()
        if w > 0
    }

    return {
        "generated_at": datetime.now(NY).isoformat(),
        "data_source": "ibkr_live" if ib is not None else "yahoo_offline",
        "fund_nav_usd": fund_nav,
        "fund_scale": fund_scale,
        "fund_weights": weights,
        "sleeve_budgets_usd": sleeve_budgets,
        "ibkr": {"connected": ib is not None, "portfolio": ib_payload},
        "signals": {
            "spy_gates": spy_gates,
            "tactical_aw": {"targets": tactical_targets, **tactical_meta},
            "vol_edge": {"targets": vol_targets, **vol_meta},
            "tsmom": {"targets": tsmom_targets, **tsmom_meta},
            "ma_slope_topn": {"targets": ma_targets, **ma_meta},
            "ma_slope_inverse": {"targets": inverse_targets, **inverse_meta},
            "johansen_etf": {"targets": johansen_targets, **johansen_meta},
            "equity_dip": dip_meta,
            "qs_actionable_etf": {"targets": qs_targets, **qs_meta},
        },
        "actions": actions,
        "bars_fetched_ib": sorted(ib_panel.keys()),
        "bars_fetched_yahoo": sorted(yahoo_panel.keys()),
    }


async def run_stock_only_signals_async(
    *,
    config_path: Path | None = None,
    fund_scale: float = 1.5,
    offline: bool = False,
) -> dict[str, Any]:
    cfg_path = config_path or (_REPO / "RenTech/live/config/live_stock_only.json")
    cfg = load_platform_config(cfg_path)
    if offline:
        return await build_stock_only_signals(None, fund_scale=fund_scale)
    ib = await connect_ib(cfg.broker)
    try:
        return await build_stock_only_signals(ib, broker=cfg.broker, fund_scale=fund_scale)
    finally:
        await disconnect_ib(ib)


def write_stock_only_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