483 lines
19 KiB
Python
483 lines
19 KiB
Python
"""实盘模拟 API 路由（spec §live-api）。
|
|
|
|
create 建 live_account(持久化配置,status=stopped);start/stop 改 status 字段;
|
|
GET 查询持仓/成交/账户/状态。runner(supervisor) 是独立常驻进程,轮询 status 字段
|
|
决定起停 LiveTradingEngine;两者只通过 DB 通信,本模块不实例化 engine。
|
|
|
|
风格参考 ``sanguo_api/routes_paper.py``。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
|
|
from fastapi import APIRouter, Depends, Header, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from .auth import verify_token as verify_token_impl
|
|
|
|
router = APIRouter()
|
|
_db_path = {"path": None}
|
|
|
|
# miniQMT 默认 userdata_mini 路径(国金QMT交易端模拟);
|
|
# req.mini_path 空 → env SANGUO_QMT_PATH → 此默认(双保险,避免 connect=-1)
|
|
_DEFAULT_MINI_PATH = r"C:\国金QMT交易端模拟\userdata_mini"
|
|
|
|
|
|
def set_db_path(p):
|
|
_db_path["path"] = p
|
|
if p:
|
|
from sanguo_live.persistence import init_db
|
|
init_db(p) # app 启动建表(幂等)
|
|
|
|
|
|
async def verify_token(authorization: str | None = Header(None)):
|
|
if authorization is None or not authorization.startswith("Bearer "):
|
|
raise HTTPException(401, "Missing/invalid authorization")
|
|
return verify_token_impl(authorization.split(" ", 1)[1])
|
|
|
|
|
|
class LiveCreateRequest(BaseModel):
|
|
name: str = "live"
|
|
account: str
|
|
vt_symbol: str = "600000"
|
|
strategy_class: str = "AShareDoubleMaStrategy"
|
|
strategy_name: str
|
|
setting: dict = {}
|
|
interval: str = "" # 空=按类型给默认(cta→15m, portfolio→d);前端下拉显式传
|
|
initial_capital: float = 1_000_000
|
|
connect_wait_sec: int = 10
|
|
init_wait_sec: int = 60
|
|
mini_path: str = ""
|
|
# 组合实盘(strategy_type='portfolio'):strategy_class 存组合策略名(all_weather 等)
|
|
strategy_type: str = "cta"
|
|
pool: str = ""
|
|
max_pool: int = 0
|
|
benchmark: str = ""
|
|
# §12.6 实例做实：账户绑档案；空=发起即建档
|
|
instance_id: int | None = None
|
|
|
|
|
|
def _resolve_file_by_class(class_name: str) -> str:
|
|
"""类名/策略名 → 策略文件名（发起即建档反查 code_file；查不到返回空）。"""
|
|
try:
|
|
from .strategy_registry import list_strategy_files
|
|
|
|
for f in list_strategy_files()["files"]:
|
|
if f.get("class_name") == class_name or f.get("name", "").removesuffix(".py") == class_name:
|
|
return f.get("name", "")
|
|
except Exception:
|
|
pass
|
|
return ""
|
|
|
|
|
|
def _normalize_vt_symbol(code: str) -> str:
|
|
"""裸 6 位码自动补交易所后缀(与 jq_to_dbbardata 同规则):
|
|
6 开头→SSE,0/3 开头→SZSE。已带后缀或非 6 位码原样返回。"""
|
|
code = (code or "").strip()
|
|
if len(code) == 6 and code.isdigit():
|
|
if code.startswith("6"):
|
|
return f"{code}.SSE"
|
|
if code.startswith(("0", "3")):
|
|
return f"{code}.SZSE"
|
|
return code
|
|
|
|
|
|
@router.post("/live/create", dependencies=[Depends(verify_token)])
|
|
def create_live(req: LiveCreateRequest):
|
|
"""创建实盘实例(写 live_accounts,status=stopped)。需调 start 才会启动。"""
|
|
from sanguo_live.persistence import init_db, save_account
|
|
|
|
db = _db_path["path"] or ":memory:"
|
|
init_db(db)
|
|
# §12.6 发起绑档案（合法 instance_id 用之；否则发起即建档）
|
|
from . import instance_store
|
|
|
|
if not (req.instance_id and instance_store.get_instance_params_snapshot(req.instance_id)):
|
|
sym = req.pool if req.strategy_type == "portfolio" else req.vt_symbol
|
|
req.instance_id = instance_store.create_instance({
|
|
"code_file": _resolve_file_by_class(req.strategy_class),
|
|
"name": f"{req.strategy_class}·live·{time.strftime('%m%d')}",
|
|
"type": req.strategy_type,
|
|
"params": dict(req.setting or {}),
|
|
"symbol_or_pool": sym,
|
|
"interval": req.interval or "d",
|
|
"match_session": "next_open",
|
|
})
|
|
else:
|
|
# D1 发起时快照：绑已有档案 → setting 用档案当时的参数（API 直调也不绕过）
|
|
req.setting = dict(instance_store.get_instance_params_snapshot(req.instance_id).get("params") or {})
|
|
# §12.6 补：发起时代码版本快照
|
|
from .code_versions import snapshot_code
|
|
|
|
inst_snap = instance_store.get_instance_params_snapshot(req.instance_id) or {}
|
|
code_snap = snapshot_code(inst_snap.get("code_file") or "")
|
|
payload = req.model_dump()
|
|
payload["code_hash"] = code_snap["code_hash"] if code_snap else None
|
|
|
|
# #78 名称服务端兜底：空/默认值 → {实例名}_v{YYYYMMDD}{minor}（同实例同日递增）。
|
|
# 前端 autoName 失败被吞时也能拿到正确名（服务端权威）。
|
|
if not payload.get("name") or payload["name"].strip() in ("", "live-600000", "live"):
|
|
inst_name = inst_snap.get("name") or "inst"
|
|
today = time.strftime("%Y%m%d")
|
|
prefix = f"{inst_name}_v{today}"
|
|
minor = 0
|
|
try:
|
|
from sanguo_live.persistence import list_accounts
|
|
|
|
ldb = _db_path["path"]
|
|
if ldb:
|
|
minor = sum(
|
|
1 for a in list_accounts(ldb)
|
|
if a.get("instance_id") == req.instance_id and (a.get("name") or "").startswith(prefix)
|
|
)
|
|
except Exception:
|
|
minor = 0
|
|
payload["name"] = f"{prefix}{minor}"
|
|
if payload.get("strategy_type") == "portfolio":
|
|
# 组合实盘:vt_symbol 占位为池名;setting 存组合参数(supervisor 转发 env)
|
|
if not payload.get("strategy_class"):
|
|
raise HTTPException(400, "组合实盘需选择策略(strategy_class)")
|
|
payload.setdefault("pool", "all")
|
|
payload.setdefault("max_pool", 30)
|
|
payload.setdefault("benchmark", "000300.XSHG")
|
|
payload["vt_symbol"] = payload["pool"]
|
|
# 周期由前端下拉传(miniQMT 成品K线档位);空=默认日线
|
|
payload["interval"] = payload.get("interval") or "d"
|
|
else:
|
|
# CTA 实盘:标的允许只写 6 位码,后端自动补交易所后缀
|
|
payload["interval"] = payload.get("interval") or "15m"
|
|
payload["vt_symbol"] = _normalize_vt_symbol(payload.get("vt_symbol", ""))
|
|
# mini_path 兜底:req → env SANGUO_QMT_PATH → 内置默认(空值会导致 connect=-1)
|
|
if not payload.get("mini_path"):
|
|
payload["mini_path"] = (
|
|
os.environ.get("SANGUO_QMT_PATH") or _DEFAULT_MINI_PATH
|
|
)
|
|
# B3 预算硬限制:Σ预算 ≤ 账户现金(fail-closed,快照不可用拒绝创建)
|
|
_enforce_budget(db, payload["account"], payload["initial_capital"])
|
|
aid = save_account(db, {**payload, "status": "stopped"})
|
|
return {"account_id": aid, "status": "stopped"}
|
|
|
|
|
|
@router.get("/live", dependencies=[Depends(verify_token)])
|
|
def list_lives():
|
|
"""实盘实例列表。每行带最新账户快照摘要(total/收益率)。
|
|
|
|
收益率用首快照基线:(last_total - first_total) / first_total,
|
|
避免用 initial_capital 兜底导致入金/出金瞬间收益率失真。
|
|
无快照时 total_return=None(不兜底 initial_capital)。
|
|
"""
|
|
from sanguo_live.persistence import (
|
|
list_accounts, get_last_balance, get_first_balance, load_positions,
|
|
)
|
|
|
|
db = _db_path["path"]
|
|
if not db:
|
|
return {"accounts": []}
|
|
items = list_accounts(db)
|
|
for item in items:
|
|
last = get_last_balance(db, item["id"])
|
|
first = get_first_balance(db, item["id"])
|
|
if last:
|
|
item["latest_equity"] = last.get("total")
|
|
item["latest_date"] = last.get("date")
|
|
else:
|
|
item["latest_equity"] = None
|
|
item["latest_date"] = None
|
|
# 收益率:首快照 total 为 baseline;last/first 同条时为 0
|
|
baseline = (first or {}).get("total") if first else None
|
|
if last and baseline:
|
|
item["total_return"] = (last.get("total", 0) - baseline) / baseline
|
|
else:
|
|
item["total_return"] = None
|
|
item["position_count"] = len(load_positions(db, item["id"]))
|
|
return {"accounts": items}
|
|
|
|
|
|
# ===== B3 预算硬限制(spec §multi-strategy-instance-budget §B3) =====
|
|
|
|
def _budget_state(
|
|
db: str, account: str, exclude_aid: int | None = None
|
|
) -> dict:
|
|
"""预算占用状态:剩余 = 账户现金 − Σ 同账户其他实盘实例预算。
|
|
|
|
fail-closed:快照缺失/过期(>10min)→ fresh=False,remaining=None(不猜数)。
|
|
Σ 不分运行/停止——停止的实例持仓仍占着账户资金。
|
|
"""
|
|
from sanguo_live.persistence import (
|
|
get_fresh_account_snapshot, list_accounts,
|
|
)
|
|
|
|
acc = (account or "").strip()
|
|
allocated = sum(
|
|
float(r.get("initial_capital") or 0)
|
|
for r in list_accounts(db)
|
|
if (r.get("account") or "").strip() == acc
|
|
and r.get("id") != exclude_aid
|
|
)
|
|
snap = get_fresh_account_snapshot(db, acc) if acc else None
|
|
if snap is None:
|
|
return {"account": acc, "fresh": False, "account_cash": None,
|
|
"allocated": allocated, "remaining": None,
|
|
"snapshot_at": None}
|
|
cash = float(snap.get("cash") or 0)
|
|
return {"account": acc, "fresh": True, "account_cash": cash,
|
|
"allocated": allocated, "remaining": cash - allocated,
|
|
"snapshot_at": snap.get("updated_at")}
|
|
|
|
|
|
def _enforce_budget(
|
|
db: str, account: str, initial_capital: float,
|
|
exclude_aid: int | None = None,
|
|
) -> dict:
|
|
"""新建/修改实盘的预算校验:超限或快照不可用 → 400(报文带剩余数)。"""
|
|
st = _budget_state(db, account, exclude_aid)
|
|
if not st["fresh"]:
|
|
raise HTTPException(
|
|
400, f"账户 {account} 快照不可用(未采集或超过10分钟),稍后再试")
|
|
if float(initial_capital) > st["remaining"]:
|
|
raise HTTPException(
|
|
400,
|
|
f"预算超限:账户可用现金 {st['account_cash']:.0f},"
|
|
f"已分配 {st['allocated']:.0f},剩余可分配 {st['remaining']:.0f}"
|
|
f"(本次 {float(initial_capital):.0f})")
|
|
return st
|
|
|
|
|
|
@router.get("/live/budget-info", dependencies=[Depends(verify_token)])
|
|
def get_budget_info(account: str = ""):
|
|
"""预算信息(前端表单默认值=remaining + 占用率条)。
|
|
|
|
注意:必须注册在 /live/{aid} 之前(FastAPI 路径匹配不按类型分流,
|
|
否则 budget-info 会被 {aid} 吃掉 422)。快照不新鲜时 fresh=False、
|
|
remaining=None——前端禁用提交并提示稍后再试(fail-closed 同款语义)。
|
|
"""
|
|
db = _db_path["path"]
|
|
if not db:
|
|
raise HTTPException(400, "db 未配置")
|
|
return _budget_state(db, account)
|
|
|
|
|
|
@router.get("/live/account-snapshot", dependencies=[Depends(verify_token)])
|
|
def get_account_snapshot_route(account: str = ""):
|
|
"""账户实况(B4 三层展示第三层):全局快照 + Σ实例市值分解。
|
|
|
|
instances = 同 QMT 账号各实盘实例的最新账本(live_balance 实例视图);
|
|
unattributed = 全账户市值 − Σ实例市值(重建后应≈0,大数=遗留仓/手动仓)。
|
|
快照缺失时 fresh=False、数值字段 None(前端显示『快照不可用』)。
|
|
"""
|
|
from sanguo_live.persistence import (
|
|
get_account_snapshot, list_accounts, get_last_balance,
|
|
)
|
|
|
|
db = _db_path["path"]
|
|
if not db:
|
|
raise HTTPException(400, "db 未配置")
|
|
acc = (account or "").strip()
|
|
snap = get_account_snapshot(db, acc) if acc else None
|
|
from sanguo_live.persistence import get_fresh_account_snapshot
|
|
fresh = get_fresh_account_snapshot(db, acc) is not None if acc else False
|
|
|
|
instances = []
|
|
for row in list_accounts(db):
|
|
if (row.get("account") or "").strip() != acc:
|
|
continue
|
|
last = get_last_balance(db, row["id"])
|
|
instances.append({
|
|
"id": row["id"], "name": row.get("name") or "",
|
|
"status": row.get("status") or "",
|
|
"initial_capital": float(row.get("initial_capital") or 0),
|
|
"cash": (last or {}).get("cash"),
|
|
"market_value": (last or {}).get("market_value"),
|
|
"equity": (last or {}).get("total"),
|
|
})
|
|
instance_mv = sum(
|
|
float(i["market_value"] or 0) for i in instances)
|
|
snap_mv = float(snap.get("market_value")) if snap else None
|
|
return {
|
|
"account": acc,
|
|
"fresh": fresh,
|
|
"cash": (snap or {}).get("cash"),
|
|
"market_value": snap_mv,
|
|
"total": (snap or {}).get("total"),
|
|
"positions": (snap or {}).get("positions") or [],
|
|
"updated_at": (snap or {}).get("updated_at"),
|
|
"instances": instances,
|
|
"instance_mv_total": instance_mv,
|
|
"unattributed_mv": (snap_mv - instance_mv
|
|
if snap_mv is not None else None),
|
|
}
|
|
|
|
|
|
@router.get("/live/{aid}", dependencies=[Depends(verify_token)])
|
|
def get_live(aid: int):
|
|
from sanguo_live.persistence import get_account, get_first_balance, get_last_balance
|
|
|
|
acc = get_account(_db_path["path"], aid)
|
|
if not acc:
|
|
raise HTTPException(404, "account not found")
|
|
# 收益率与列表页同口径:首快照为基线。监控页此前用 initial_capital 兜底,
|
|
# 共用 QMT 账户时 total=1000万 vs cap=100万 → 假 900%(2026-08-14 实况)。
|
|
last = get_last_balance(_db_path["path"], aid)
|
|
first = get_first_balance(_db_path["path"], aid)
|
|
baseline = (first or {}).get("total") if first else None
|
|
acc["latest_equity"] = (last or {}).get("total") if last else None
|
|
acc["latest_date"] = (last or {}).get("date") if last else None
|
|
if last and baseline:
|
|
acc["total_return"] = (last.get("total", 0) - baseline) / baseline
|
|
else:
|
|
acc["total_return"] = None
|
|
return acc
|
|
|
|
|
|
@router.post("/live/{aid}/start", dependencies=[Depends(verify_token)])
|
|
def start_live(aid: int):
|
|
"""启动实例(status=running)。supervisor 轮询发现后起 engine。"""
|
|
from sanguo_live.persistence import get_account, update_account_status
|
|
|
|
acc = get_account(_db_path["path"], aid)
|
|
if not acc:
|
|
raise HTTPException(404, "account not found")
|
|
if not acc["account"]:
|
|
raise HTTPException(400, "account 字段(交易账号)不能为空")
|
|
update_account_status(_db_path["path"], aid, "running")
|
|
return {"account_id": aid, "status": "running"}
|
|
|
|
|
|
@router.post("/live/{aid}/stop", dependencies=[Depends(verify_token)])
|
|
def stop_live(aid: int):
|
|
"""停止实例(status=stopped)。supervisor 轮询发现后停 engine。"""
|
|
from sanguo_live.persistence import get_account, update_account_status
|
|
|
|
if not get_account(_db_path["path"], aid):
|
|
raise HTTPException(404, "account not found")
|
|
update_account_status(_db_path["path"], aid, "stopped")
|
|
return {"account_id": aid, "status": "stopped"}
|
|
|
|
|
|
@router.get("/live/{aid}/positions", dependencies=[Depends(verify_token)])
|
|
def get_positions(aid: int):
|
|
"""持仓快照(读 live_positions,supervisor 定时落库)。"""
|
|
from sanguo_live.persistence import load_positions
|
|
|
|
return load_positions(_db_path["path"], aid)
|
|
|
|
|
|
@router.get("/live/{aid}/trades", dependencies=[Depends(verify_token)])
|
|
def get_trades(aid: int):
|
|
"""成交明细(读 live_trades,supervisor 事件回调落库)。"""
|
|
from sanguo_live.persistence import list_trades
|
|
|
|
return list_trades(_db_path["path"], aid)
|
|
|
|
|
|
@router.get("/live/{aid}/account", dependencies=[Depends(verify_token)])
|
|
def get_account_balance(aid: int):
|
|
"""账户最新快照(读 live_balance 最新一条)。"""
|
|
from sanguo_live.persistence import get_last_balance
|
|
|
|
last = get_last_balance(_db_path["path"], aid)
|
|
return last or {}
|
|
|
|
|
|
@router.get("/live/{aid}/status", dependencies=[Depends(verify_token)])
|
|
def get_status(aid: int):
|
|
"""运行状态(读 live_accounts.status)。"""
|
|
from sanguo_live.persistence import get_account
|
|
|
|
acc = get_account(_db_path["path"], aid)
|
|
if not acc:
|
|
raise HTTPException(404, "account not found")
|
|
return {
|
|
"account_id": aid, "status": acc["status"], "name": acc["name"],
|
|
"account": acc["account"], "vt_symbol": acc["vt_symbol"],
|
|
"strategy_name": acc["strategy_name"], "updated_at": acc["updated_at"],
|
|
"error_msg": acc.get("error_msg", ""),
|
|
}
|
|
|
|
|
|
# ===== 生命周期管理补全：删除/编辑（停止/启动已有）=====
|
|
|
|
class LiveUpdateRequest(BaseModel):
|
|
"""可编辑字段。仅 stopped 状态可改（运行中改配置会与 engine 失配）。"""
|
|
name: str | None = None
|
|
account: str | None = None
|
|
vt_symbol: str | None = None
|
|
strategy_class: str | None = None
|
|
strategy_name: str | None = None
|
|
setting: dict | None = None
|
|
interval: str | None = None
|
|
# B3:预算可改(initial_capital 含义升级为资金额度)
|
|
initial_capital: float | None = None
|
|
|
|
|
|
@router.put("/live/{aid}", dependencies=[Depends(verify_token)])
|
|
def update_live(aid: int, req: LiveUpdateRequest):
|
|
import json
|
|
import sqlite3
|
|
|
|
from sanguo_live.persistence import get_account
|
|
|
|
db = _db_path["path"]
|
|
acc = get_account(db, aid)
|
|
if not acc:
|
|
raise HTTPException(404, "account not found")
|
|
if acc["status"] == "running":
|
|
raise HTTPException(400, "运行中不可编辑，请先停止实例")
|
|
# B3:预算或账号变更时校验(改名等不触发——不是预算事件,夜间 QMT 关闭
|
|
# 也能改名)。exclude 自身:自己的旧预算不重复计入 Σ。
|
|
if req.initial_capital is not None or (
|
|
req.account and req.account != acc["account"]
|
|
):
|
|
_enforce_budget(
|
|
db,
|
|
req.account if req.account is not None else acc["account"],
|
|
req.initial_capital if req.initial_capital is not None
|
|
else float(acc.get("initial_capital") or 0),
|
|
exclude_aid=aid,
|
|
)
|
|
sets, args = [], []
|
|
field_map = {
|
|
"name": req.name, "account": req.account, "vt_symbol": req.vt_symbol,
|
|
"strategy_class": req.strategy_class, "strategy_name": req.strategy_name,
|
|
"interval": req.interval,
|
|
}
|
|
for col, v in field_map.items():
|
|
if v is not None:
|
|
# 编辑与新建同规:裸 6 位码补交易所后缀(2026-08-14 实况:编辑漏补
|
|
# → 引擎 "vt_symbol 无法解析,跳过" → 假运行收不到行情)
|
|
if col == "vt_symbol":
|
|
v = _normalize_vt_symbol(v)
|
|
sets.append(f"{col}=?"); args.append(v)
|
|
if req.setting is not None:
|
|
sets.append("setting=?"); args.append(json.dumps(req.setting))
|
|
if req.initial_capital is not None:
|
|
sets.append("initial_capital=?"); args.append(req.initial_capital)
|
|
if not sets:
|
|
return {"account_id": aid, "updated": False}
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute(f"UPDATE live_accounts SET {', '.join(sets)} WHERE id=?", (*args, aid))
|
|
conn.commit()
|
|
return {"account_id": aid, "updated": True}
|
|
|
|
|
|
@router.delete("/live/{aid}", dependencies=[Depends(verify_token)])
|
|
def delete_live(aid: int):
|
|
"""删除实盘实例及其数据。运行中拒绝删除（先 stop）。"""
|
|
import sqlite3
|
|
|
|
from sanguo_live.persistence import get_account
|
|
|
|
db = _db_path["path"]
|
|
acc = get_account(db, aid)
|
|
if not acc:
|
|
raise HTTPException(404, "account not found")
|
|
if acc["status"] == "running":
|
|
raise HTTPException(400, "运行中不可删除，请先停止实例")
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute("DELETE FROM live_accounts WHERE id=?", (aid,))
|
|
for t in ("live_trades", "live_positions", "live_balance"):
|
|
conn.execute(f"DELETE FROM {t} WHERE account_id=?", (aid,))
|
|
conn.commit()
|
|
return {"account_id": aid, "deleted": True}
|