65 lines
2.5 KiB
Python
65 lines
2.5 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"
|