96 lines
4.1 KiB
Python
96 lines
4.1 KiB
Python
"""影子柜台主管(supervisor)纯逻辑测试:账户筛选 + env 映射。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from sanguo_trader.persistence import init_db, save_account
|
|
from sanguo_trader.shadow.supervisor import account_env, load_shadow_accounts
|
|
|
|
|
|
def _mk_account(**kw) -> dict:
|
|
base = dict(
|
|
name="shadow-acc", mode="shadow", strategy_type="portfolio",
|
|
interval="15m", symbols=["hs300_subset"],
|
|
strategies=[{"name": "channel_test", "params": {"max_pool": 6,
|
|
"benchmark": "000905.XSHG"}}],
|
|
initial_capital=500_000, rate=0.00025, stamp_duty_rate=0.001,
|
|
min_commission=5, slippage=0.002,
|
|
start="2026-08-14", end="2026-12-31", engine="shadow",
|
|
)
|
|
base.update(kw)
|
|
return base
|
|
|
|
|
|
def test_load_shadow_accounts_filters(tmp_path):
|
|
db = str(tmp_path / "p.db")
|
|
init_db(db)
|
|
from sanguo_trader.persistence import update_account_status
|
|
a_shadow = save_account(db, _mk_account()) # 应选中
|
|
update_account_status(db, a_shadow, "running") # 创建后 API 置 running
|
|
a_stopped = save_account(db, _mk_account(name="stopped", strategies=[
|
|
{"name": "x", "params": {}}]))
|
|
update_account_status(db, a_stopped, "stopped") # 停止 → 不选
|
|
a_live = save_account(db, _mk_account(name="live", mode="live")) # 实走 → 不选
|
|
_ = a_live
|
|
|
|
accounts = load_shadow_accounts(db)
|
|
assert [a["id"] for a in accounts] == [a_shadow]
|
|
|
|
|
|
def test_account_env_mapping(tmp_path):
|
|
db = str(tmp_path / "p.db")
|
|
init_db(db)
|
|
aid = save_account(db, _mk_account())
|
|
from sanguo_trader.persistence import update_account_status
|
|
update_account_status(db, aid, "running")
|
|
acc = load_shadow_accounts(db)[0]
|
|
env = account_env(acc, db)
|
|
assert env["SANGUO_LIVE_STRATEGY"] == "channel_test"
|
|
assert env["SANGUO_LIVE_MAX_POOL"] == "6"
|
|
assert env["SANGUO_LIVE_BENCHMARK"] == "000905.XSHG"
|
|
assert env["SANGUO_LIVE_CASH"] == "500000"
|
|
assert env["SANGUO_SHADOW_DB"] == db
|
|
assert env["SANGUO_SHADOW_ACCOUNT_ID"] == str(aid)
|
|
assert env["SANGUO_SHADOW_COMMISSION"] == "0.00025"
|
|
assert env["SANGUO_SHADOW_SLIPPAGE"] == "0.002"
|
|
|
|
|
|
def test_account_env_defaults_on_sparse_row():
|
|
acc = {"id": 9, "strategies": json.dumps([]),
|
|
"initial_capital": None, "rate": None}
|
|
env = account_env(acc, "db")
|
|
assert env["SANGUO_LIVE_STRATEGY"] == "all_weather"
|
|
assert env["SANGUO_LIVE_CASH"] == "1000000"
|
|
assert env["SANGUO_SHADOW_ACCOUNT_ID"] == "9"
|
|
|
|
|
|
def test_account_env_factor_params_passthrough():
|
|
"""#69 影子A/B 挑战者:params factor 五键 → SANGUO_LIVE_FACTOR_* env 齐透传。"""
|
|
acc = {"id": 43, "strategies": json.dumps([{"name": "factor_topn", "params": {
|
|
"factor_name": "challenger_v1",
|
|
"factor_dir": r"C:\sanguo_vnpy_v2\data\factors\challenger",
|
|
"top_n": 200.0,
|
|
"reb_days": 21,
|
|
"band": 1.5,
|
|
}}]), "initial_capital": 1_000_000}
|
|
env = account_env(acc, "db")
|
|
assert env["SANGUO_LIVE_FACTOR_NAME"] == "challenger_v1"
|
|
# factor_dir 字符串原样透传
|
|
assert env["SANGUO_LIVE_FACTOR_DIR"] == r"C:\sanguo_vnpy_v2\data\factors\challenger"
|
|
assert env["SANGUO_LIVE_FACTOR_TOP_N"] == "200" # 整数不带小数点(200.0 → "200")
|
|
assert env["SANGUO_LIVE_FACTOR_REB_DAYS"] == "21"
|
|
assert env["SANGUO_LIVE_FACTOR_BAND"] == "1.5"
|
|
|
|
|
|
def test_account_env_factor_absent_not_set(monkeypatch):
|
|
"""params 无 factor 键:不设 SANGUO_LIVE_FACTOR_*(live_strategy 默认值兜底)。"""
|
|
for k in ("SANGUO_LIVE_FACTOR_NAME", "SANGUO_LIVE_FACTOR_DIR",
|
|
"SANGUO_LIVE_FACTOR_TOP_N", "SANGUO_LIVE_FACTOR_REB_DAYS",
|
|
"SANGUO_LIVE_FACTOR_BAND"):
|
|
monkeypatch.delenv(k, raising=False)
|
|
acc = {"id": 44, "strategies": json.dumps([{"name": "factor_topn",
|
|
"params": {"max_pool": 0}}]),
|
|
"initial_capital": None}
|
|
env = account_env(acc, "db")
|
|
assert not [k for k in env if k.startswith("SANGUO_LIVE_FACTOR_")]
|