"""影子柜台常驻进程入口（P1，VPS Windows / miniQMT 行情）。

与组合实盘(``sanguo_portfolio.runner_live``)同一个 bullet_trade LiveEngine，
唯一区别:broker_factory 换成 ShadowBroker(本地撮合,订单不出门)。
策略/行情/调度完全同款 → 双轨一致性验证(设计 §8)的基础。

环境变量(复用 live_strategy.py 的 SANGUO_LIVE_* 命名 + 影子专属 SANGUO_SHADOW_*):
    SANGUO_LIVE_STRATEGY/_MAX_POOL/_BENCHMARK/_CASH   策略配置(live_strategy.py 读)
    SANGUO_SHADOW_DB / SANGUO_SHADOW_ACCOUNT_ID       落库目标(paper 库)
    SANGUO_SHADOW_COMMISSION/_STAMP/_MIN_COMM/_SLIPPAGE  费率滑点(对齐实盘券商参数)

手动用法(VPS 交易日):
    set SANGUO_LIVE_STRATEGY=all_weather
    set SANGUO_SHADOW_DB=C:\\sanguo_vnpy_v2\\data\\paper.db
    python -m sanguo_trader.shadow

不做多账户轮询:MVP 一进程一账户(与 runner_live 一致),多账户由 supervisor
按 paper_accounts(engine='shadow')逐行拉子进程(后续接入)。
"""
from __future__ import annotations

# ENV GUARD 必须早于任何 bullet_trade import
import os
os.environ.setdefault("DEFAULT_DATA_PROVIDER", "miniqmt")

import logging
import threading
import time
from pathlib import Path
from typing import Any, Dict, Optional

logger = logging.getLogger(__name__)

# 策略适配文件与组合实盘共用(读 SANGUO_LIVE_* env)
ADAPTER_FILE = Path(__file__).resolve().parents[2] / "sanguo_portfolio" / "live_strategy.py"


def shadow_env() -> Dict[str, str]:
    """解析影子柜台 env(独立出来便于单测)。"""
    return {
        "db": os.environ.get("SANGUO_SHADOW_DB", ""),
        "account_id": os.environ.get("SANGUO_SHADOW_ACCOUNT_ID", ""),
        "commission": os.environ.get("SANGUO_SHADOW_COMMISSION", "0.0003"),
        "stamp": os.environ.get("SANGUO_SHADOW_STAMP", "0.001"),
        "min_comm": os.environ.get("SANGUO_SHADOW_MIN_COMM", "5"),
        "slippage": os.environ.get("SANGUO_SHADOW_SLIPPAGE", "0.001"),
        "snapshot_sec": os.environ.get("SANGUO_SHADOW_SNAPSHOT_SEC", "30"),
    }


def build_price_getter(provider: Any) -> Any:
    """从数据 provider 取标的最新价(实时/最新收盘)。返回闭包给 ShadowBroker。"""

    def get_price(security: str) -> Optional[float]:
        from datetime import datetime, timedelta

        end = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        start = (datetime.now() - timedelta(days=10)).strftime("%Y-%m-%d")
        try:
            df = provider.get_price(
                security=security, start_date=start, end_date=end,
                frequency="daily", fields=["close"], fq="pre",
            )
            if df is None or len(df) == 0:
                return None
            return float(df["close"].iloc[-1])
        except Exception:  # noqa: BLE001 - provider 接口差异兜底
            cols = [c for c in ("close", "Close") if c in (df.columns if df is not None else [])]
            if cols:
                return float(df[cols[0]].iloc[-1])
            return None

    return get_price


def build_limit_getter(provider: Any) -> Any:
    """从数据 provider 取标的实时涨跌停/停牌状态(P1.3,ShadowBroker 拒单用)。

    优先 ``get_live_current``(miniQMT 实时 tick:lastPrice vs UpStop/DownStop,
    与实盘同源同刻);无实时能力(回测 provider)→ 回退 ``get_limit_status_batch``
    (日线 prev_close×幅度口径)。查不到 → None(ShadowBroker 放行,等价旧行为)。
    """
    from datetime import datetime
    from typing import Optional as _Opt

    def get_limit(security: str) -> _Opt[dict]:
        live_fn = getattr(provider, "get_live_current", None)
        if live_fn is not None:
            try:
                cur = live_fn(security) or {}
                last = cur.get("last_price")
                high = cur.get("high_limit") or 0.0
                low = cur.get("low_limit") or 0.0
                if last:
                    return {
                        "is_limit_up": bool(high and float(last) >= float(high)),
                        "is_limit_down": bool(low and float(last) <= float(low)),
                        "is_paused": bool(cur.get("paused")),
                    }
            except Exception:  # noqa: BLE001 - 实时源失败试批量源
                pass
        batch_fn = getattr(provider, "get_limit_status_batch", None)
        if batch_fn is not None:
            try:
                return (batch_fn([security], datetime.now().strftime("%Y-%m-%d"))
                        or {}).get(security)
            except Exception:  # noqa: BLE001
                return None
        return None

    return get_limit


def _paper_on_trade(db: str, account_id: int, strategy_id: str):
    """成交回调:落 paper_trades(与组合实走 EOD 同表,前端模拟盘页直接可见)。"""
    from sanguo_trader.persistence import save_trade

    def hook(trade: Dict[str, Any]) -> None:
        side = trade["side"]
        save_trade(db, account_id, {
            "strategy_id": strategy_id,
            "datetime": trade["datetime"],
            "symbol": trade["security"],
            "direction": "long" if side == "buy" else "short",
            "offset": "open" if side == "buy" else "close",
            "match_session": "shadow_realtime",
            "price": trade["price"],
            "volume": trade["amount"],
            "commission": trade["commission"],
            "stamp_duty": trade["stamp_duty"],
            "bar_date": trade["datetime"][:10],
        })

    return hook


def _should_write_balance(as_of: str, last_written: str) -> bool:
    """#88:有行情日期且与上次已写日期不同才写(同日一行;空日期=无行情不写)。

    原实现每 30s 无条件写一行,VPS 实况 40 号账户半天刷 97 行 date='' 垃圾
    (周末无行情 as_of='',净值恒 100 万),把「日净值表」刷成流水账。
    """
    return bool(as_of) and as_of != last_written


def _snapshot_loop(broker: Any, db: str, account_id: int,
                   interval_sec: float = 30.0) -> None:
    """后台线程:定期把影子账户快照落 paper_positions/paper_daily_balance。"""
    from sanguo_trader.persistence import save_daily_balance, save_positions

    last_as_of = ""
    while True:
        time.sleep(interval_sec)
        try:
            info = broker.get_account_info()
            positions = {
                p["security"]: {"volume": float(p["amount"]), "frozen": 0.0,
                                "avg_price": p["avg_cost"]}
                for p in info["positions"]
            }
            save_positions(db, account_id, "account", positions,
                           date=broker.trades[-1]["datetime"][:10] if broker.trades else "")
            as_of = info.get("as_of", "")
            if _should_write_balance(as_of, last_as_of):
                save_daily_balance(
                    db, account_id, as_of,
                    cash=info["available_cash"], market_value=info["market_value"],
                    total_equity=info["total_value"],
                )
                last_as_of = as_of
        except Exception as exc:  # noqa: BLE001 - 落库失败不中断柜台
            logger.warning("[shadow-snapshot] 落库失败 (account=%s): %s", account_id, exc)


def _hist_trades_for_restore(rows: list) -> list:
    """paper_trades 行 → broker.restore_from_trades 入参(重放用交易字典)。

    独立成函数便于单测:offset=open→buy / close→sell 是恢复链路的唯一契约。
    """
    return [
        {"security": r["symbol"],
         "side": "buy" if r["offset"] == "open" else "sell",
         "amount": int(r["volume"]), "price": float(r["price"]),
         "commission": float(r["commission"] or 0),
         "stamp_duty": float(r["stamp_duty"] or 0),
         "datetime": r["datetime"]}
        for r in rows
    ]


def _instance_adapter(account_id: str) -> Path:
    """按账户复制一份策略适配文件(同 runner_live._instance_adapter)。

    bullet_trade 实例锁按 strategy_path+broker_type+account_identity 判重;
    多影子账户共用 ShadowBroker(同 identity)时须路径互异才不被判重复实例。
    """
    if not account_id:
        return ADAPTER_FILE
    dst = (Path(__file__).resolve().parents[2] / "runtime"
           / f"shadow_{account_id}" / ADAPTER_FILE.name)
    dst.parent.mkdir(parents=True, exist_ok=True)
    if not dst.exists() or dst.read_text(encoding="utf-8") != \
            ADAPTER_FILE.read_text(encoding="utf-8"):
        import shutil
        shutil.copyfile(ADAPTER_FILE, dst)
    return dst


def run_shadow(provider_config: Optional[Dict[str, Any]] = None) -> None:
    """装配 LiveEngine(影子 broker)并 run(阻塞)。"""
    from bullet_trade.core.live_engine import LiveEngine  # type: ignore
    from bullet_trade.data.api import set_data_provider  # type: ignore

    from sanguo_portfolio.runner_live import build_provider
    from .broker import ShadowBroker

    cfg = shadow_env()
    from sanguo_portfolio.runner_live import live_env
    le = live_env()

    provider = build_provider(provider_config)
    set_data_provider(provider)

    broker = ShadowBroker(
        initial_cash=float(le["cash"]),
        commission_rate=float(cfg["commission"]),
        stamp_duty_rate=float(cfg["stamp"]),
        min_commission=float(cfg["min_comm"]),
        slippage=float(cfg["slippage"]),
        price_getter=build_price_getter(provider),
        limit_getter=build_limit_getter(provider),
        on_trade=_paper_on_trade(cfg["db"], int(cfg["account_id"]), le["strategy"])
        if cfg["db"] and cfg["account_id"] else None,
    )
    # 重启恢复:账本全在内存,不重放 paper_trades 就会重置回初始资金/0 持仓,
    # 与已落库成交断层(2026-08-19 前 VPS 实况:重启即"失忆")
    if cfg["db"] and cfg["account_id"]:
        from sanguo_trader.persistence import list_trades

        hist = _hist_trades_for_restore(
            list_trades(cfg["db"], int(cfg["account_id"])))
        if hist:
            broker.restore_from_trades(hist)
            logger.info("[shadow] 重启恢复 %d 笔历史成交: cash=%.2f 持仓 %d 只",
                        len(hist), broker.cash, len(broker.positions))
    logger.info(
        "影子柜台启动: strategy=%s cash=%s 费率=佣金%s/印花%s/最低%s 滑点%s db=%s",
        le["strategy"], le["cash"], cfg["commission"], cfg["stamp"],
        cfg["min_comm"], cfg["slippage"], cfg["db"] or "(不落库)",
    )

    engine = LiveEngine(
        _instance_adapter(cfg["account_id"]),
        broker_factory=lambda: broker,
        # 独立 runtime 目录:bullet_trade 单实例锁默认 ./runtime,
        # 影子与实盘并行(双轨§8)会互抢锁,按账户分目录隔离
        live_config={"runtime_dir": str(
            Path(__file__).resolve().parent.parent.parent / "runtime"
            / f"shadow_{cfg['account_id'] or 'solo'}")},
    )

    if cfg["db"] and cfg["account_id"]:
        t = threading.Thread(
            target=_snapshot_loop,
            args=(broker, cfg["db"], int(cfg["account_id"]), float(cfg["snapshot_sec"])),
            daemon=True, name="shadow-snapshot",
        )
        t.start()

    engine.run()