sanguo_vnpy_v2/tests/api/test_portfolio_live.py

"""Tests for 组合策略实盘(R3-1): live_accounts 组合列 + create 路由 + supervisor env 映射。

runner_live 的引擎装配依赖 miniQMT/VPS,不在单测范围;这里测的是
「web 建组合实盘 → DB 字段落对 → supervisor 能翻译出正确 env」这条契约链。
直调路由函数风格与 test_paper_lifecycle.py 一致(免 auth)。
"""
import sqlite3

import pytest

from sanguo_api import routes_live as rl
from sanguo_live import persistence as live_persistence
from sanguo_live import runner as live_runner


@pytest.fixture()
def live_db(tmp_path):
    db = str(tmp_path / "live.db")
    rl.set_db_path(db)
    return db


def _create_portfolio(db, **kw):
    req = rl.LiveCreateRequest(
        name=kw.get("name", "组合实盘1"), account=kw.get("account", "66639661"),
        strategy_name=kw.get("strategy_name", "portfolio_all_weather"),
        strategy_type="portfolio", strategy_class=kw.get("strategy_class", "all_weather"),
        pool=kw.get("pool", "hs300_subset"), max_pool=kw.get("max_pool", 30),
        benchmark=kw.get("benchmark", "000300.XSHG"),
        interval=kw.get("interval", ""),
        initial_capital=kw.get("initial_capital", 500000),
    )
    return rl.create_live(req)["account_id"]


def test_save_account_portfolio_fields(live_db):
    aid = live_persistence.save_account(live_db, {
        "name": "portfolio-live", "account": "66639661",
        "strategy_class": "all_weather", "strategy_name": "portfolio_all_weather",
        "strategy_type": "portfolio", "pool": "hs300_subset",
        "max_pool": 30, "benchmark": "000300.XSHG",
        "status": "running", "vt_symbol": "hs300_subset",
    })
    acc = live_persistence.get_account(live_db, aid)
    assert acc["strategy_type"] == "portfolio"
    assert acc["pool"] == "hs300_subset"
    assert acc["max_pool"] == 30
    assert acc["benchmark"] == "000300.XSHG"


def test_init_db_migrates_old_live_accounts(tmp_path):
    """旧库(无组合列)init_db 后补列且默认 cta。"""
    db = str(tmp_path / "old.db")
    old_schema = """
    CREATE TABLE live_accounts (
        id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, account TEXT,
        vt_symbol TEXT, strategy_class TEXT, strategy_name TEXT, setting TEXT,
        status TEXT, interval TEXT, initial_capital REAL,
        connect_wait_sec INTEGER, init_wait_sec INTEGER, mini_path TEXT,
        error_msg TEXT, created_at TEXT, updated_at TEXT
    );
    """
    with sqlite3.connect(db) as conn:
        conn.executescript(old_schema)
        conn.execute("INSERT INTO live_accounts (name, status) VALUES ('old', 'stopped')")
    live_persistence.init_db(db)
    acc = live_persistence.get_account(db, 1)
    assert acc["strategy_type"] == "cta"  # ALTER DEFAULT 生效于旧行


def test_create_portfolio_live(live_db):
    aid = _create_portfolio(live_db)
    acc = live_persistence.get_account(live_db, aid)
    assert acc["strategy_type"] == "portfolio"
    assert acc["vt_symbol"] == "hs300_subset"  # 组合行 vt_symbol=池名
    assert acc["interval"] == "d"  # 未传周期 → 组合默认日线
    assert acc["status"] == "stopped"


def test_create_portfolio_interval_passthrough(live_db):
    """组合实盘周期由前端下拉传(miniQMT 档位),后端不写死。"""
    aid = _create_portfolio(live_db, interval="15m")
    acc = live_persistence.get_account(live_db, aid)
    assert acc["interval"] == "15m"


def test_create_portfolio_rejects_empty_strategy(live_db):
    with pytest.raises(Exception):
        _create_portfolio(live_db, strategy_class="")


def test_create_cta_defaults_unchanged(live_db):
    req = rl.LiveCreateRequest(account="66639661", strategy_name="dm1")
    aid = rl.create_live(req)["account_id"]
    acc = live_persistence.get_account(live_db, aid)
    assert acc["strategy_type"] == "cta"
    assert acc["interval"] == "15m"
    assert acc["vt_symbol"] == "600000.SSE"


def test_portfolio_env_mapping():
    """live_accounts 行 → runner_live env(env 是组合实盘唯一参数契约)。"""
    acc = {
        "id": 7, "account": "66639661", "mini_path": r"C:\qmt\userdata_mini",
        "strategy_class": "value_selection", "max_pool": 20,
        "benchmark": "000905.XSHG", "initial_capital": 2_000_000,
    }
    env = live_runner._portfolio_env_for(acc, "live.db")
    assert env["SANGUO_QMT_ACCOUNT"] == "66639661"
    assert env["SANGUO_QMT_PATH"] == r"C:\qmt\userdata_mini"
    assert env["SANGUO_LIVE_STRATEGY"] == "value_selection"
    assert env["SANGUO_LIVE_MAX_POOL"] == "20"
    assert env["SANGUO_LIVE_BENCHMARK"] == "000905.XSHG"
    assert env["SANGUO_LIVE_CASH"] == "2000000"
    assert env["SANGUO_LIVE_DB"] == "live.db"
    assert env["SANGUO_LIVE_ACCOUNT_ID"] == "7"


def test_runner_live_env_defaults(monkeypatch):
    """runner_live.live_env 带默认值(手动跑不传参也不炸)。"""
    from sanguo_portfolio import runner_live
    for k in ("SANGUO_LIVE_STRATEGY", "SANGUO_LIVE_MAX_POOL", "SANGUO_LIVE_BENCHMARK",
              "SANGUO_LIVE_CASH", "SANGUO_QMT_ACCOUNT", "SANGUO_QMT_PATH",
              "SANGUO_LIVE_DB", "SANGUO_LIVE_ACCOUNT_ID"):
        monkeypatch.delenv(k, raising=False)
    cfg = runner_live.live_env()
    assert cfg["strategy"] == "all_weather"
    assert cfg["max_pool"] == "30"
    assert cfg["benchmark"] == "000300.XSHG"
    assert cfg["account"] == ""  # 空 → run_live 拒绝启动(防误下单)


def test_live_strategy_adapter_builds_all_strategies(monkeypatch):
    """适配文件的策略工厂:env → StrategyTemplate(4 策略各识别一次)。"""
    from sanguo_portfolio import live_strategy

    class _FakeProvider:
        pass

    for name, cls_name in (
        ("all_weather", "AllWeatherStrategy"),
        ("momentum_timing", "MomentumTimingStrategy"),
        ("value_selection", "ValueSelectionStrategy"),
        ("small_cap", "SmallCapStrategy"),
        ("channel_test", "ChannelTestStrategy"),
        # TET Phase2 副本:影子/实盘同样可发起(2026-08-16 VPS shadow#42 因缺项拉起即崩)
        ("all_weather_ex", "AllWeatherExStrategy"),
        ("momentum_timing_ex", "MomentumTimingExStrategy"),
        ("value_selection_ex", "ValueSelectionExStrategy"),
        ("small_cap_ex", "SmallCapExStrategy"),
    ):
        monkeypatch.setenv("SANGUO_LIVE_STRATEGY", name)
        s = live_strategy._build_live_strategy(_FakeProvider())
        assert type(s).__name__ == cls_name
    monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "nope")
    with pytest.raises(ValueError):
        live_strategy._build_live_strategy(_FakeProvider())


def _patch_live_wiring(monkeypatch):
    """打桩 bullet_trade 顶层 API:记录定时注册,provider 给假件。"""
    import bullet_trade.core as bt_core
    import bullet_trade.data.api as bt_data_api

    registered = []
    monkeypatch.setattr(
        bt_core, "run_daily",
        lambda f, t, **kw: registered.append(
            ("daily", getattr(f, "__name__", str(f)), t)))
    monkeypatch.setattr(
        bt_core, "run_monthly",
        lambda f, d, t, **kw: registered.append(
            ("monthly", getattr(f, "__name__", str(f)), d, t)))

    class _FakeProvider:
        pass

    monkeypatch.setattr(bt_data_api, "get_data_provider", lambda: _FakeProvider())
    return registered


def _reset_live_state():
    from sanguo_portfolio import live_strategy
    live_strategy._STATE["strategy"] = None
    live_strategy._STATE["wired"] = False


def test_process_initialize_registers_tasks_on_resume_boot(monkeypatch):
    """P0 回归:resume 重启(g 已恢复,引擎跳过 initialize 只调 process_initialize)
    定时任务仍要注册——2026-08-17 VPS 16 引擎重启后零任务空转、开盘零成交。"""
    from sanguo_portfolio import live_strategy

    registered = _patch_live_wiring(monkeypatch)
    monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "all_weather")
    _reset_live_state()

    # 只调 process_initialize,不调 initialize(复刻 resume 路径)
    live_strategy.process_initialize(object())

    names = {r[1] for r in registered}
    assert "prepare_stock_list" in names
    assert "monthly_adjustment" in names
    assert "stop_loss" in names
    assert live_strategy._STATE["wired"] is True


def test_setup_idempotent_across_both_hooks(monkeypatch):
    """新策略首启:引擎先调 initialize 再调 process_initialize——任务不重复注册,
    策略实例复用同一份。"""
    from sanguo_portfolio import live_strategy

    registered = _patch_live_wiring(monkeypatch)
    monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "momentum_timing")
    _reset_live_state()

    live_strategy.initialize(object())
    first = live_strategy._STATE["strategy"]
    live_strategy.process_initialize(object())

    assert live_strategy._STATE["strategy"] is first
    assert len([r for r in registered if r[1] == "handle_data"]) == 1


def test_facade_injects_run_daily_for_channel_test(monkeypatch):
    """channel_test 自注册依赖 facade.run_daily(缺注入时静默 no-op 永不开仓):
    process_initialize 后 9:35/10:45/13:45/14:30 四时点应经注入的 run_daily 挂上。"""
    import bullet_trade.core as bt_core
    from sanguo_portfolio import live_strategy

    registered = _patch_live_wiring(monkeypatch)
    monkeypatch.setenv("SANGUO_LIVE_STRATEGY", "channel_test")
    _reset_live_state()

    live_strategy.process_initialize(object())

    daily_times = sorted(r[2] for r in registered if r[0] == "daily")
    assert daily_times == ["10:45", "13:45", "14:30", "9:35"]
    # facade 拿到的是 bullet_trade 顶层 run_daily 本尊,不是默认空 lambda
    assert live_strategy._STATE["strategy"].broker.run_daily is bt_core.run_daily


def test_normalize_vt_symbol_bare_code_gets_exchange_suffix():
    """CTA 实盘标的:裸 6 位码自动补交易所后缀(6→SSE,0/3→SZSE)。"""
    from sanguo_api.routes_live import _normalize_vt_symbol

    assert _normalize_vt_symbol("600000") == "600000.SSE"
    assert _normalize_vt_symbol("000001") == "000001.SZSE"
    assert _normalize_vt_symbol("300750") == "300750.SZSE"
    # 已带后缀/非 6 位(池名等)原样
    assert _normalize_vt_symbol("600000.SSE") == "600000.SSE"
    assert _normalize_vt_symbol("hs300_subset") == "hs300_subset"
    assert _normalize_vt_symbol("") == ""


def test_update_live_normalizes_vt_symbol(tmp_path, monkeypatch):
    """编辑实盘裸码补后缀(2026-08-14 实况:编辑漏补→引擎订阅跳过→假运行)。"""
    import os
    from fastapi.testclient import TestClient
    from sanguo_api.app import create_app
    from sanguo_api.auth import create_token, set_jwt_config
    from sanguo_api.routes_live import set_db_path

    set_jwt_config(secret="t", expire_minutes=60)
    db = os.path.join(str(tmp_path), "l.db")
    app = create_app(db_path=db)
    set_db_path(db)
    c = TestClient(app)
    h = {"Authorization": f"Bearer {create_token('admin')}"}

    aid = c.post("/api/v1/live/create", json={
        "name": "t", "account": "A1", "vt_symbol": "600000.SSE",
        "strategy_class": "AShareDoubleMaStrategy",
        "strategy_name": "AShareDoubleMa_600000",
    }, headers=h).json()["account_id"]
    r = c.put(f"/api/v1/live/{aid}", json={"vt_symbol": "300024"}, headers=h)
    assert r.status_code == 200
    acc = c.get(f"/api/v1/live/{aid}", headers=h).json()
    assert acc["vt_symbol"] == "300024.SZSE"


def test_snapshot_once_skips_unsynced_cash():
    """cash<=0(账户未同步完成)不落 balance——治假收益率(2026-08-14 实况 341080%)。"""
    import os
    import tempfile
    import types

    from sanguo_live.persistence import init_db, list_balance
    from sanguo_portfolio.runner_live import _snapshot_once

    db = os.path.join(tempfile.mkdtemp(), "l.db")
    init_db(db)

    def _mk_portfolio(cash, total):
        p = types.SimpleNamespace(
            available_cash=cash, total_value=total, positions={})
        ctx = types.SimpleNamespace(portfolio=p)
        return types.SimpleNamespace(context=ctx)

    # 未同步完成: cash=0, total=持仓市值 → 不落 balance
    _snapshot_once(_mk_portfolio(0, 2931.0), db, 3)
    assert list_balance(db, 3) == []
    # 正常: cash>0 → 落库
    _snapshot_once(_mk_portfolio(9_997_077.51, 10_000_008.51), db, 3)
    rows = list_balance(db, 3)
    assert len(rows) == 1
    assert rows[0]["total"] == 10_000_008.51


def test_get_live_return_uses_first_snapshot_baseline(live_db):
    """监控页收益率与列表页同口径(首快照基线),不再用 initial_capital 兜底。"""
    from sanguo_live.persistence import save_balance
    aid = _create_portfolio(live_db)
    save_balance(live_db, aid, "2026-08-14 20:00:00", 9_000_000, 1_000_000,
                 total=10_000_000)
    save_balance(live_db, aid, "2026-08-14 21:00:00", 9_100_000, 1_050_000,
                 total=10_150_000)
    acc = rl.get_live(aid)
    assert acc["latest_equity"] == 10_150_000
    assert acc["total_return"] == (10_150_000 - 10_000_000) / 10_000_000