diff --git a/config/data_platform.yaml b/config/data_platform.yaml index 51e6919..829d657 100644 --- a/config/data_platform.yaml +++ b/config/data_platform.yaml @@ -49,3 +49,12 @@ live: mode_b: false # D-4c 模式B: bridge回报校正账本（默认关，切实盘再开） # 和 Windows bridge 同值；不进 git。占位空值，真实值部署时填实际 config bridge_token: + +# 实盘模拟(task #4)— supervisor 常驻进程 + API 共享 DB +# db_path 留空则 fallback 到 data_paths.vnpy_db(与回测主库同) +# supervisor 用法: python -m sanguo_live --supervisor [db_path] +live_trading: + enabled: false # 总开关 + db_path: # 留空 → 用 data_paths.vnpy_db + poll_interval_sec: 5 # supervisor 轮询 live_accounts.status 间隔 + snapshot_interval_sec: 30 # 持仓/账户快照落库间隔 diff --git a/config/live.yaml b/config/live.yaml new file mode 100644 index 0000000..f7dee45 --- /dev/null +++ b/config/live.yaml @@ -0,0 +1,34 @@ +# 实盘模拟交易配置 +# 用法: +# python -m sanguo_live +# SANGUO_QMT_ACCOUNT=66639661 python -m sanguo_live +# +# env SANGUO_QMT_ACCOUNT / SANGUO_QMT_PATH 优先于此文件。 + +# miniQMT 交易账号(可用 env SANGUO_QMT_ACCOUNT 覆盖) +account: "66639661" + +# userdata_mini 路径;留空则由 vnpy_qmt/md.py 自动扫描 C:\ +# (避免中文路径字面量编码问题,推荐留空或用 env SANGUO_QMT_PATH) +mini_path: "" + +# 策略实例名(唯一,用于 CTA 引擎路由) +strategy_name: "dm_15min_600000" + +# 策略类名(必须在 sanguo_live.runner._STRATEGY_REGISTRY 注册) +strategy_class: "AShareDoubleMaStrategy" + +# 标的 vt_symbol(SYMBOL.EXCHANGE)。600000.SSE = 浦发银行 +vt_symbol: "600000.SSE" + +# 各阶段等待秒数 +connect_wait_sec: 10 +init_wait_sec: 60 + +# 策略参数(透传给 CtaTemplate.update_setting) +setting: + fast_window: 10 + slow_window: 20 + window: 15 # BarGenerator 分钟窗口(A 股 15min) + size: 100 # 1 手 = 100 股 + forbid_short: true # A 股不可做空 → short() 拦截 diff --git a/frontend/src/api/live.ts b/frontend/src/api/live.ts new file mode 100644 index 0000000..7a303bb --- /dev/null +++ b/frontend/src/api/live.ts @@ -0,0 +1,125 @@ +import { apiClient } from './client' + +/** 实盘模拟账户(API 返回行) */ +export interface LiveAccount { + id: number + name: string + account: string + vt_symbol: string + strategy_class: string + strategy_name: string + /** JSON 字符串,前端 JSON.parse 得到策略参数 */ + setting: string + status: string + interval: string + initial_capital: number + connect_wait_sec?: number + init_wait_sec?: number + mini_path?: string + error_msg?: string | null + created_at?: string + updated_at?: string + /** 列表端点附带(单账户 GET 不含) */ + latest_equity?: number | null + latest_date?: string | null + total_return?: number | null + position_count?: number +} + +export interface LiveCreateRequest { + name: string + account: string + vt_symbol: string + strategy_class: string + strategy_name: string + setting: Record + interval: string + initial_capital: number + connect_wait_sec?: number + init_wait_sec?: number + mini_path?: string +} + +export interface LiveStatus { + account_id: number + status: string + name: string + account: string + vt_symbol: string + strategy_name: string + updated_at?: string + error_msg?: string +} + +export interface LivePosition { + symbol: string + volume: number + frozen: number + avg_price: number + updated_at?: string +} + +export interface LiveTrade { + account_id: number + strategy_name: string + symbol: string + direction: string + offset: string + price: number + volume: number + traded_at: string + vt_tradeid?: string +} + +export interface LiveBalance { + account_id?: number + date?: string + cash?: number + market_value?: number + total?: number +} + +export async function createLive(req: LiveCreateRequest): Promise<{ accountId: number; status: string }> { + const { data } = await apiClient.post<{ account_id: number; status: string }>('/live/create', req) + return { accountId: data.account_id, status: data.status } +} + +export async function listLives(): Promise { + const { data } = await apiClient.get<{ accounts: LiveAccount[] }>('/live') + return data.accounts +} + +export async function getLive(aid: number): Promise { + const { data } = await apiClient.get(`/live/${aid}`) + return data +} + +export async function startLive(aid: number): Promise<{ accountId: number; status: string }> { + const { data } = await apiClient.post<{ account_id: number; status: string }>(`/live/${aid}/start`) + return { accountId: data.account_id, status: data.status } +} + +export async function stopLive(aid: number): Promise<{ accountId: number; status: string }> { + const { data } = await apiClient.post<{ account_id: number; status: string }>(`/live/${aid}/stop`) + return { accountId: data.account_id, status: data.status } +} + +export async function getLivePositions(aid: number): Promise { + const { data } = await apiClient.get(`/live/${aid}/positions`) + return data +} + +export async function getLiveTrades(aid: number): Promise { + const { data } = await apiClient.get(`/live/${aid}/trades`) + return data +} + +export async function getLiveAccountBalance(aid: number): Promise { + const { data } = await apiClient.get(`/live/${aid}/account`) + return data ?? {} +} + +export async function getLiveStatus(aid: number): Promise { + const { data } = await apiClient.get(`/live/${aid}/status`) + return data +} diff --git a/frontend/src/api/portfolio.ts b/frontend/src/api/portfolio.ts new file mode 100644 index 0000000..5215477 --- /dev/null +++ b/frontend/src/api/portfolio.ts @@ -0,0 +1,68 @@ +import { apiClient } from './client' + +export interface PortfolioBacktestReq { + pool: string + start_date: string + end_date: string + initial_cash: number + benchmark?: string +} + +export interface EquityPoint { + date: string + equity: number +} + +export interface StockPicked { + code: string + name: string + amount: number + avg_cost: number + price: number + value: number +} + +export interface PortfolioTrade { + datetime?: string + date?: string + code?: string + side?: string + action?: string + amount?: number + filled_amount?: number + price?: number + filled_price?: number + commission?: number + status?: string +} + +export interface PortfolioMetrics { + total_return: number | null + annual_return: number | null + max_drawdown: number | null + sharpe: number | null + win_rate_daily: number | null + win_rate_trade: number | null + trading_days: number | null +} + +export interface PortfolioBacktestResult { + strategy: string + period: { start: string; end: string; trading_days: number } + stocks_selected: StockPicked[] + trades: PortfolioTrade[] + equity_curve: EquityPoint[] + metrics: PortfolioMetrics + raw_summary?: Record +} + +export async function postPortfolioBacktest( + req: PortfolioBacktestReq, +): Promise { + const { data } = await apiClient.post( + '/portfolio/backtest', + req, + { timeout: 600000 }, + ) + return data +} diff --git a/frontend/src/router/index.ts b/frontend/src/router/index.ts index 026ecfc..8329626 100644 --- a/frontend/src/router/index.ts +++ b/frontend/src/router/index.ts @@ -10,6 +10,7 @@ const routes: RouteRecordRaw[] = [ { path: '', redirect: '/dashboard' }, { path: 'dashboard', name: 'dashboard', component: () => import('@/views/Dashboard.vue') }, { path: 'backtest/new', name: 'bt-new', component: () => import('@/views/backtest/New.vue') }, + { path: 'backtest/portfolio', name: 'bt-portfolio', component: () => import('@/views/backtest/PortfolioBacktest.vue') }, { path: 'backtest/progress/:id', name: 'bt-progress', component: () => import('@/views/backtest/Progress.vue') }, { path: 'backtest/result/:id', name: 'bt-result', component: () => import('@/views/backtest/Result.vue') }, { path: 'backtest/optimize', name: 'bt-optimize', component: () => import('@/views/backtest/Optimize.vue') }, @@ -22,6 +23,9 @@ const routes: RouteRecordRaw[] = [ { path: 'paper', name: 'paper-list', component: () => import('@/views/paper/List.vue') }, { path: 'paper/result/:id', name: 'paper-result', component: () => import('@/views/paper/Result.vue') }, { path: 'paper/live/:aid', name: 'paper-live', component: () => import('@/views/paper/Live.vue') }, + { path: 'live/new', name: 'live-new', component: () => import('@/views/live/New.vue') }, + { path: 'live', name: 'live-list', component: () => import('@/views/live/List.vue') }, + { path: 'live/monitor/:id', name: 'live-monitor', component: () => import('@/views/live/Monitor.vue') }, ], }, ] diff --git a/frontend/src/views/Layout.vue b/frontend/src/views/Layout.vue index dda3e4a..ab41589 100644 --- a/frontend/src/views/Layout.vue +++ b/frontend/src/views/Layout.vue @@ -18,6 +18,7 @@ const activeMenu = computed(() => { const PAGE_TITLE: Array<{ match: RegExp; group: string; title: string }> = [ { match: /^\/dashboard$/, group: '工作台', title: '工作台' }, { match: /^\/backtest\/new$/, group: '回测', title: '新建回测' }, + { match: /^\/backtest\/portfolio$/, group: '回测', title: '组合回测' }, { match: /^\/backtest\/optimize$/, group: '回测', title: '参数优化' }, { match: /^\/backtest\/history$/, group: '回测', title: '历史任务' }, { match: /^\/backtest\/progress\//, group: '回测', title: '任务进度' }, @@ -29,6 +30,9 @@ const PAGE_TITLE: Array<{ match: RegExp; group: string; title: string }> = [ { match: /^\/paper\/result\//, group: '模拟', title: '模拟盘结果' }, { match: /^\/paper\/live\//, group: '模拟', title: '实走监控' }, { match: /^\/paper$/, group: '模拟', title: '模拟交易' }, + { match: /^\/live\/new$/, group: '实盘模拟', title: '新建实盘' }, + { match: /^\/live\/monitor\//, group: '实盘模拟', title: '实盘监控' }, + { match: /^\/live$/, group: '实盘模拟', title: '实盘模拟' }, ] const pageMeta = computed(() => { const p = route.path @@ -59,6 +63,7 @@ function onLogout(): void { 回测 新建回测 + 组合回测 参数优化 历史任务 @@ -78,9 +83,13 @@ function onLogout(): void { 新建模拟盘 - - 实盘 国金 QMT · D 期 - + + + 实盘列表 + 新建实盘 +
v2 · 量化研究台
diff --git a/frontend/src/views/backtest/PortfolioBacktest.vue b/frontend/src/views/backtest/PortfolioBacktest.vue new file mode 100644 index 0000000..2ec20cf --- /dev/null +++ b/frontend/src/views/backtest/PortfolioBacktest.vue @@ -0,0 +1,319 @@ + + + + + diff --git a/frontend/src/views/live/List.vue b/frontend/src/views/live/List.vue new file mode 100644 index 0000000..16243d9 --- /dev/null +++ b/frontend/src/views/live/List.vue @@ -0,0 +1,213 @@ + + + + + diff --git a/frontend/src/views/live/Monitor.vue b/frontend/src/views/live/Monitor.vue new file mode 100644 index 0000000..95162bc --- /dev/null +++ b/frontend/src/views/live/Monitor.vue @@ -0,0 +1,333 @@ + + + + + diff --git a/frontend/src/views/live/New.vue b/frontend/src/views/live/New.vue new file mode 100644 index 0000000..147475c --- /dev/null +++ b/frontend/src/views/live/New.vue @@ -0,0 +1,180 @@ + + + + + diff --git a/pytest.ini b/pytest.ini index b5fee15..533a891 100644 --- a/pytest.ini +++ b/pytest.ini @@ -2,3 +2,5 @@ pythonpath = . testpaths = tests asyncio_mode = auto +markers = + requires_bullet_trade: 需要 bullet-trade 已安装(否则 skip) diff --git a/sanguo_api/app.py b/sanguo_api/app.py index 1bfd654..a67b569 100644 --- a/sanguo_api/app.py +++ b/sanguo_api/app.py @@ -3,7 +3,9 @@ FastAPI application factory for Sanguo Quant API """ from fastapi import FastAPI from .routes import router, set_orchestrator, set_auth_config -from .routes_paper import router as paper_router, set_db_path +from .routes_paper import router as paper_router, set_db_path as set_paper_db_path +from .routes_live import router as live_router, set_db_path as set_live_db_path +from .routes_portfolio import router as portfolio_router from .auth import set_jwt_config from .ws import manager from sanguo_orchestrator.runner import Orchestrator @@ -35,7 +37,10 @@ def create_app(db_path: str, file_dir=None, auth_config=None, max_workers: int = # Include routes app.include_router(router, prefix="/api/v1") app.include_router(paper_router, prefix="/api/v1") - set_db_path(db_path) + app.include_router(live_router, prefix="/api/v1") + app.include_router(portfolio_router, prefix="/api/v1") + set_paper_db_path(db_path) + set_live_db_path(db_path) @app.on_event("startup") def _register_live_step(): diff --git a/sanguo_api/routes_live.py b/sanguo_api/routes_live.py new file mode 100644 index 0000000..1686af2 --- /dev/null +++ b/sanguo_api/routes_live.py @@ -0,0 +1,178 @@ +"""实盘模拟 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 + +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.SSE" + strategy_class: str = "AShareDoubleMaStrategy" + strategy_name: str + setting: dict = {} + interval: str = "15m" + initial_capital: float = 1_000_000 + connect_wait_sec: int = 10 + init_wait_sec: int = 60 + mini_path: str = "" + + +@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) + payload = req.model_dump() + # 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 + ) + 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} + + +@router.get("/live/{aid}", dependencies=[Depends(verify_token)]) +def get_live(aid: int): + from sanguo_live.persistence import get_account + + acc = get_account(_db_path["path"], aid) + if not acc: + raise HTTPException(404, "account not found") + 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", ""), + } diff --git a/sanguo_api/routes_portfolio.py b/sanguo_api/routes_portfolio.py new file mode 100644 index 0000000..616d32d --- /dev/null +++ b/sanguo_api/routes_portfolio.py @@ -0,0 +1,126 @@ +"""组合策略 API 路由(MVP)。 + +POST /portfolio/backtest: SSH 触发 VPS 跑 BulletTrade + all_weather, +捕获 stdout JSON 返回前端。同步模式(回测耗时,前端 loading,timeout 600s)。 + +风格参考 routes_paper.py / routes_live.py。 +""" +from __future__ import annotations + +import logging +import shlex +import subprocess +from typing import Any, Optional + +from fastapi import APIRouter, Depends, Header, HTTPException +from pydantic import BaseModel, Field + +from .auth import verify_token as verify_token_impl + +logger = logging.getLogger(__name__) +router = APIRouter() + +# VPS SSH 连接配置(~/.ssh/config 已配 49.232.102.198 免密 key) +_VPS_HOST = "49.232.102.198" +_VPS_WORKDIR = r"C:\\sanguo_vnpy_v2" +_VPS_PYTHON = "python" +_VPS_TIMEOUT = 600 # 回测耗时,给 10 分钟 + + +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 PortfolioBacktestRequest(BaseModel): + pool: str = Field(default="hs300_subset", description="标的池(占位,MVP 用默认)") + start_date: str = Field(default="2024-01-01", description="YYYY-MM-DD") + end_date: str = Field(default="2024-02-29", description="YYYY-MM-DD") + initial_cash: float = Field(default=1_000_000.0, description="初始资金(元)") + benchmark: str = Field(default="000300.XSHG", description="基准代码") + + +@router.post("/portfolio/backtest", dependencies=[Depends(verify_token)]) +def run_portfolio_backtest(req: PortfolioBacktestRequest): + """SSH 触发 VPS 跑 BulletTrade + all_weather 回测,同步返回 JSON 结果。 + + Windows SSH 坑: + - GBK 编码:python -X utf8 避免中文 print 编码崩 + - 引号嵌套:用 list 形式 argv,避免 shell 引号 escape 噩梦 + - 没有 tail/head/grep:用 python 后处理(本函数在 Mac 端直接解析 stdout) + - 中文路径:VPS_WORKDIR / userdata_mini 走 env(DEFAULT_DATA_PROVIDER=miniqmt) + """ + # 在 VPS 上跑的命令:cd workdir && set ENV && python -m sanguo_portfolio.runner_backtest --json + # Windows cmd: set X=Y&&cmd2 (注意 & 必须紧贴前一条,不能有空格,否则 set 会把尾部空格算进 value) + cmd_parts = [ + "set", "DEFAULT_DATA_PROVIDER=miniqmt", "&&", + "cd", _VPS_WORKDIR, "&&", + _VPS_PYTHON, "-X", "utf8", "-m", "sanguo_portfolio.runner_backtest", + "--json", + "--start", req.start_date, + "--end", req.end_date, + "--cash", str(req.initial_cash), + "--benchmark", req.benchmark, + ] + # 用 ssh host "cmd string" 形式;argv 在 ssh 远端走 cmd /c 解析 + ssh_argv = [ + "ssh", + "-o", "ConnectTimeout=15", + "-o", "StrictHostKeyChecking=no", + _VPS_HOST, + " ".join(shlex.quote(p) if p != "&&" else "&&" for p in cmd_parts), + ] + + logger.info("[portfolio] SSH 触发: %s", ssh_argv[-1]) + try: + proc = subprocess.run( + ssh_argv, + capture_output=True, + text=True, + timeout=_VPS_TIMEOUT, + check=False, + ) + except subprocess.TimeoutExpired: + raise HTTPException(504, f"VPS 回测超时(>{_VPS_TIMEOUT}s)") + except FileNotFoundError: + raise HTTPException(500, "本机未找到 ssh 命令") + except Exception as exc: + logger.exception("[portfolio] SSH 调用失败") + raise HTTPException(500, f"SSH 调用失败: {exc}") + + if proc.returncode != 0: + stderr_tail = (proc.stderr or "")[-2000:] + logger.error("[portfolio] VPS 回测失败 rc=%s stderr=%s", proc.returncode, stderr_tail) + raise HTTPException( + 500, + f"VPS 回测失败(rc={proc.returncode}): {stderr_tail}", + ) + + # 从 stdout 提取最后一行 JSON(runner --json 只 print 一行) + stdout = proc.stdout or "" + import json as _json + result: Optional[dict[str, Any]] = None + parse_err: Optional[str] = None + for line in reversed(stdout.strip().splitlines()): + line = line.strip() + if not line.startswith("{"): + continue + try: + result = _json.loads(line) + break + except _json.JSONDecodeError as exc: + parse_err = str(exc) + continue + + if result is None: + logger.error( + "[portfolio] stdout 无 JSON 行。parse_err=%s stdout_tail=%s", + parse_err, stdout[-2000:], + ) + raise HTTPException( + 500, + f"VPS stdout 解析失败: {parse_err or 'no JSON line'}; stdout_tail={stdout[-500:]!r}", + ) + + return result diff --git a/sanguo_live/__init__.py b/sanguo_live/__init__.py new file mode 100644 index 0000000..1525702 --- /dev/null +++ b/sanguo_live/__init__.py @@ -0,0 +1,14 @@ +"""sanguo_live — vnpy 原生实盘交易引擎(A股 15min 策略)。 + +模块组成: +- base_template.AShareCtaTemplate A股 CTA 基类(定寸 size=100 + 禁做空) +- strategies.AShareDoubleMaStrategy 15min 双均线策略 +- engine.LiveTradingEngine MainEngine + QmtGateway + CtaStrategyApp 封装 +- runner.run 实盘入口: connect → add → subscribe → init → start → 常驻 + +本机 dev 视图通常未装 vnpy_ctastrategy / vnpy_qmt,模块 import 容错(运行时才校验依赖)。 +""" +from __future__ import annotations + +__all__ = ["__version__"] +__version__ = "0.1.0" diff --git a/sanguo_live/__main__.py b/sanguo_live/__main__.py new file mode 100644 index 0000000..2239fc8 --- /dev/null +++ b/sanguo_live/__main__.py @@ -0,0 +1,17 @@ +"""python -m sanguo_live [config_path | --supervisor [db_path]] + +默认(无参 / yaml 路径):config 驱动单实例模式(run)。 +--supervisor [db_path]:DB 驱动 supervisor,轮询 live_accounts.status 管理多实例。 +""" +import sys + +from sanguo_live.runner import run, run_supervisor + +if __name__ == "__main__": + args = sys.argv[1:] + if args and args[0] == "--supervisor": + db_path = args[1] if len(args) > 1 else None + run_supervisor(db_path) + else: + cfg = args[0] if args else None + run(cfg) diff --git a/sanguo_live/base_template.py b/sanguo_live/base_template.py new file mode 100644 index 0000000..7c5ffad --- /dev/null +++ b/sanguo_live/base_template.py @@ -0,0 +1,117 @@ +"""A股 CTA 策略基类:定寸(1 手 = 100 股)+ 禁做空。 + +为什么需要这一层: +- vnpy_qmt 的 send_order 把 ``volume`` 当**股数**直接传给 xtquant,不会自动乘合约 size。 + 标准 CtaTemplate 策略写 ``buy(price, 1)`` 表示 1 手 → 实际只下 1 股,会被券商拒单 + 或当成废单。AShareCtaTemplate 在 buy/sell/cover 内部把 volume ×= ``self.size`` + (默认 100),让策略保留"手"的语义。 +- A 股不可做空:``short`` 直接返回 [] 并写日志(``forbid_short=True`` 生效时)。 + +依赖 vnpy_ctastrategy.CtaTemplate。本机未装时 CtaTemplate fallback 为 ``object``, +模块 import 不崩,只有运行时实例化或调用方法才会报错。 +""" +from __future__ import annotations + +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +try: + from vnpy_ctastrategy import CtaTemplate # type: ignore + _HAS_CTA_BASE = True + _IMPORT_ERROR: Exception | None = None +except ImportError as _e: # 本机 dev 环境常未装 vnpy_ctastrategy + CtaTemplate = object # type: ignore[assignment,misc] + _HAS_CTA_BASE = False + _IMPORT_ERROR = _e + + +class AShareCtaTemplate(CtaTemplate): # type: ignore[misc] + """A 股 CTA 策略基类。 + + 定寸:``buy`` / ``cover`` / ``sell`` 的 ``volume`` 自动 ``×= size``(默认 100, + 即 1 手 = 100 股),不足 1 手向下取整并告警。 + 禁做空:``short`` 直接返回 [](由 ``forbid_short`` 控制,默认 True)。 + + 子类化注意:必须把 ``size``/``forbid_short`` 加入自己的 ``parameters`` 列表, + 否则 ``update_setting`` 不会回填这两个字段。 + """ + + # 1 手 = 100 股(A 股最小交易单位)。ContractData.size 也是 100。 + size: int = 100 + # True: short() 被拦截(A 股不能开空);False: 透传到基类(仅供测试/期货场景)。 + forbid_short: bool = True + + parameters = ["size", "forbid_short"] + + def buy( + self, + price: float, + volume: float, + stop: bool = False, + lock: bool = False, + net: bool = False, + ) -> list: + """开多 → A 股买入,定寸到整手。""" + return super().buy(price, self._to_lots(volume), stop, lock, net) + + def cover( + self, + price: float, + volume: float, + stop: bool = False, + lock: bool = False, + net: bool = False, + ) -> list: + """平空 → A 股 Normally 不会触达(禁做空);保留定寸以防策略逻辑误调。""" + return super().cover(price, self._to_lots(volume), stop, lock, net) + + def sell( + self, + price: float, + volume: float, + stop: bool = False, + lock: bool = False, + net: bool = False, + ) -> list: + """平多 → A 股卖出持仓,定寸到整手。""" + return super().sell(price, self._to_lots(volume), stop, lock, net) + + def short( + self, + price: float, + volume: float, + stop: bool = False, + lock: bool = False, + net: bool = False, + ) -> list: + """开空 → A 股不可做空,默认拦截。""" + if self.forbid_short: + self.write_log( + f"A股禁做空: 拦截 short price={price} volume={volume}" + ) + return [] + return super().short(price, self._to_lots(volume), stop, lock, net) + + def _to_lots(self, volume: float) -> int: + """策略手数 → A 股股数(``volume × size``,不足 1 手向下取整)。""" + target = volume * self.size + lots = int(target) + if lots != target: + logger.warning( + "%s 定寸出现零股: volume=%s size=%s → %d (向下取整)", + getattr(self, "strategy_name", "?"), volume, self.size, lots, + ) + return lots + + +def ensure_base_available() -> None: + """显式检查 vnpy_ctastrategy 是否就绪。模块加载时容错,真正实盘前调一次。""" + if not _HAS_CTA_BASE: + raise RuntimeError( + f"vnpy_ctastrategy 未安装,无法实例化 AShareCtaTemplate: {_IMPORT_ERROR}" + ) + + +__all__ = ["AShareCtaTemplate", "ensure_base_available"] diff --git a/sanguo_live/engine.py b/sanguo_live/engine.py new file mode 100644 index 0000000..1ac9972 --- /dev/null +++ b/sanguo_live/engine.py @@ -0,0 +1,154 @@ +"""LiveTradingEngine:vnpy 原生实盘链路封装。 + +参考 ``vnpy_v4.4.0/examples/no_ui/run.py`` 和 ``examples/no_ui/run.py``, +按 A 股 + QMT 场景精简: +- EventEngine + MainEngine +- add_gateway(QmtGateway) 连 miniQMT +- add_app(CtaStrategyApp) 挂 CTA 引擎 +- 透出 connect / add_strategy / subscribe / init_all / start_all / stop_all / 查询 + +本机未装 vnpy_ctastrategy / vnpy_qmt 时 import 容错,实例化才报错。 +""" +from __future__ import annotations + +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +try: + from vnpy.event import EventEngine # type: ignore + from vnpy.trader.engine import MainEngine # type: ignore + from vnpy_qmt import QmtGateway # type: ignore + from vnpy_ctastrategy import CtaStrategyApp, CtaEngine # type: ignore + _DEPS_OK = True + _IMPORT_ERROR: Exception | None = None +except ImportError as _e: + EventEngine = None # type: ignore + MainEngine = None # type: ignore + QmtGateway = None # type: ignore + CtaStrategyApp = None # type: ignore + CtaEngine = None # type: ignore + _DEPS_OK = False + _IMPORT_ERROR = _e + + +class LiveTradingEngine: + """vnpy 原生实盘引擎封装。单实例持有 MainEngine 生命周期。 + + 使用: + eng = LiveTradingEngine() + eng.connect({"交易账号": "66639661", "mini路径": "C:\\\\..."}) + eng.add_strategy(AShareDoubleMaStrategy, "dm1", "600000.SSE", {...}) + eng.subscribe(["600000.SSE"]) + eng.init_all(); eng.start_all() + # ... 常驻 ... + eng.stop_all(); eng.close() + """ + + def __init__(self) -> None: + if not _DEPS_OK: + raise RuntimeError( + f"vnpy 依赖缺失(vnpy/vnpy_qmt/vnpy_ctastrategy)," + f"无法初始化 LiveTradingEngine: {_IMPORT_ERROR}" + ) + self.event_engine: EventEngine = EventEngine() + self.main_engine: MainEngine = MainEngine(self.event_engine) + self.main_engine.add_gateway(QmtGateway) + self.cta_engine: CtaEngine = self.main_engine.add_app(CtaStrategyApp) + logger.info("LiveTradingEngine 初始化完成(MainEngine + QMT + CTA)") + + # ---------------------- 连接 / 策略 / 行情 ---------------------- + + def connect(self, setting: dict[str, str]) -> None: + """连接 miniQMT。``setting`` = ``{"交易账号":..., "mini路径":...}``。""" + self.main_engine.connect(setting, "QMT") + logger.info("已请求连接 QMT: 账号=%s", setting.get("交易账号")) + + def add_strategy( + self, + strategy_class: type, + strategy_name: str, + vt_symbol: str, + setting: dict[str, Any], + ) -> None: + """注册策略实例到 CTA 引擎。""" + self.cta_engine.add_strategy( + strategy_class, strategy_name, vt_symbol, setting + ) + logger.info("已添加策略 %s @ %s (class=%s)", strategy_name, vt_symbol, + strategy_class.__name__) + + def subscribe(self, vt_symbols: list[str]) -> None: + """订阅 ``vt_symbol`` 列表(格式 ``SYMBOL.EXCHANGE``,如 ``600000.SSE``)。 + + 订阅是行情驱动策略的前提:QmtGateway.on_tick → EVENT_TICK → + CtaEngine → strategy.on_tick → BarGenerator 合成 15min bar。 + """ + from vnpy.trader.object import SubscribeRequest # type: ignore + from vnpy.trader.constant import Exchange # type: ignore + + ok = 0 + for vt_symbol in vt_symbols: + try: + symbol, exch_code = vt_symbol.split(".", 1) + exchange = Exchange(exch_code) + except (ValueError, KeyError): + logger.warning("vt_symbol 无法解析,跳过: %r", vt_symbol) + continue + req = SubscribeRequest(symbol=symbol, exchange=exchange) + self.main_engine.subscribe(req, "QMT") + ok += 1 + logger.info("已请求订阅 %d / %d 个标的", ok, len(vt_symbols)) + + # ---------------------- 生命周期 ---------------------- + + def init_all(self) -> None: + self.cta_engine.init_all_strategies() + logger.info("所有策略初始化完成") + + def start_all(self) -> None: + self.cta_engine.start_all_strategies() + logger.info("所有策略已启动") + + def stop_all(self) -> None: + try: + self.cta_engine.stop_all_strategies() + logger.info("所有策略已停止") + except Exception as e: # noqa: BLE001 + logger.warning("stop_all_strategies 异常: %s", e) + + def close(self) -> None: + try: + self.main_engine.close() + finally: + logger.info("LiveTradingEngine 已关闭") + + # ---------------------- 查询(OMS 缓存) ---------------------- + + def _oms(self) -> Any: + return self.main_engine.get_engine("oms") + + def get_all_accounts(self) -> list: + """返回 QMT 网关所有账户(AccountData 列表)。连接前可能为空。""" + oms = self._oms() + if oms is None: + return [] + return [a for a in oms.get_all_accounts() if a.gateway_name == "QMT"] + + def get_positions(self) -> list: + """返回 QMT 持仓(PositionData 列表)。""" + oms = self._oms() + if oms is None: + return [] + return [p for p in oms.get_all_positions() if p.gateway_name == "QMT"] + + def get_orders(self) -> list: + """返回 QMT 当日委托(OrderData 列表)。""" + oms = self._oms() + if oms is None: + return [] + return [o for o in oms.get_all_orders() if o.gateway_name == "QMT"] + + +__all__ = ["LiveTradingEngine"] diff --git a/sanguo_live/persistence.py b/sanguo_live/persistence.py new file mode 100644 index 0000000..23b2c05 --- /dev/null +++ b/sanguo_live/persistence.py @@ -0,0 +1,280 @@ +"""实盘模拟 SQLite 持久化(4 表,spec §live-api)。 + +表：live_accounts / live_trades / live_positions / live_balance +WAL 模式支持 supervisor 进程写 + API 进程读（DB 解耦,spec §live-api）。 + +设计参考 ``sanguo_trader/persistence.py``（paper_* 表），但： +- account.setting 存 JSON 字符串（策略参数透传给 CtaTemplate.update_setting） +- status: stopped | running（API 改字段,supervisor 轮询该字段决定起停） +- positions 为覆盖式快照（supervisor 定时把 OMS PositionData 落库,不做增量） +""" +from __future__ import annotations + +import json +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS 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 +); +CREATE TABLE IF NOT EXISTS live_trades ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + account_id INTEGER, + strategy_name TEXT, + symbol TEXT, + direction TEXT, + offset TEXT, + price REAL, + volume REAL, + traded_at TEXT, + vt_tradeid TEXT +); +CREATE TABLE IF NOT EXISTS live_positions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + account_id INTEGER, + symbol TEXT, + volume REAL, + frozen REAL, + avg_price REAL, + updated_at TEXT, + UNIQUE(account_id, symbol) +); +CREATE TABLE IF NOT EXISTS live_balance ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + account_id INTEGER, + date TEXT, + cash REAL, + market_value REAL, + total REAL +); +CREATE INDEX IF NOT EXISTS idx_live_trades_account ON live_trades(account_id); +CREATE INDEX IF NOT EXISTS idx_live_balance_account ON live_balance(account_id, date); +""" + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def init_db(db_path: str) -> None: + Path(db_path).parent.mkdir(parents=True, exist_ok=True) + with sqlite3.connect(db_path) as conn: + conn.executescript(SCHEMA) + conn.execute("PRAGMA journal_mode=WAL") + conn.commit() + + +# ----------------- live_accounts CRUD ----------------- + +def save_account(db_path: str, account: dict[str, Any]) -> int: + with sqlite3.connect(db_path) as conn: + cur = conn.execute( + """INSERT INTO live_accounts + (name, account, vt_symbol, strategy_class, strategy_name, setting, + status, interval, initial_capital, connect_wait_sec, init_wait_sec, + mini_path, created_at, updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + account.get("name", "live"), + account.get("account", ""), + account.get("vt_symbol", ""), + account.get("strategy_class", "AShareDoubleMaStrategy"), + account.get("strategy_name", ""), + json.dumps(account.get("setting", {})), + account.get("status", "stopped"), + account.get("interval", "15m"), + account.get("initial_capital", 1_000_000), + int(account.get("connect_wait_sec", 10)), + int(account.get("init_wait_sec", 60)), + account.get("mini_path", ""), + _now(), _now(), + ), + ) + conn.commit() + return cur.lastrowid + + +def list_accounts(db_path: str) -> list[dict]: + with sqlite3.connect(db_path) as conn: + conn.row_factory = sqlite3.Row + cur = conn.execute("SELECT * FROM live_accounts ORDER BY id DESC") + return [dict(r) for r in cur.fetchall()] + + +def get_account(db_path: str, account_id: int) -> dict | None: + with sqlite3.connect(db_path) as conn: + conn.row_factory = sqlite3.Row + cur = conn.execute( + "SELECT * FROM live_accounts WHERE id=?", (account_id,) + ) + row = cur.fetchone() + return dict(row) if row else None + + +def update_account_status( + db_path: str, account_id: int, status: str, error_msg: str = "" +) -> None: + with sqlite3.connect(db_path) as conn: + conn.execute( + "UPDATE live_accounts SET status=?, error_msg=?, updated_at=? WHERE id=?", + (status, error_msg, _now(), account_id), + ) + conn.commit() + + +def list_running_accounts(db_path: str) -> list[dict]: + """supervisor 轮询：取所有 status=running 的实例。""" + with sqlite3.connect(db_path) as conn: + conn.row_factory = sqlite3.Row + cur = conn.execute( + "SELECT * FROM live_accounts WHERE status=? ORDER BY id", ("running",) + ) + return [dict(r) for r in cur.fetchall()] + + +# ----------------- live_trades ----------------- + +def save_trade(db_path: str, account_id: int, trade: dict[str, Any]) -> int: + with sqlite3.connect(db_path) as conn: + cur = conn.execute( + """INSERT INTO live_trades + (account_id, strategy_name, symbol, direction, offset, + price, volume, traded_at, vt_tradeid) + VALUES (?,?,?,?,?,?,?,?,?)""", + ( + account_id, trade.get("strategy_name", ""), + trade.get("symbol", ""), trade.get("direction", ""), + trade.get("offset", ""), trade.get("price", 0), + trade.get("volume", 0), trade.get("traded_at", ""), + trade.get("vt_tradeid", ""), + ), + ) + conn.commit() + return cur.lastrowid + + +def list_trades(db_path: str, account_id: int) -> list[dict]: + with sqlite3.connect(db_path) as conn: + conn.row_factory = sqlite3.Row + cur = conn.execute( + "SELECT * FROM live_trades WHERE account_id=? ORDER BY id", + (account_id,), + ) + return [dict(r) for r in cur.fetchall()] + + +# ----------------- live_positions (覆盖式快照) ----------------- + +def save_positions( + db_path: str, account_id: int, positions: dict[str, dict] +) -> None: + """覆盖式落库。positions = {symbol: {volume, frozen, avg_price}}。 + + supervisor 每 snapshot_interval_sec 调一次,把 OMS 最新 PositionData 覆盖落库。 + 只保留 volume>0 的持仓。 + """ + now = _now() + with sqlite3.connect(db_path) as conn: + conn.execute("DELETE FROM live_positions WHERE account_id=?", (account_id,)) + conn.executemany( + """INSERT INTO live_positions + (account_id, symbol, volume, frozen, avg_price, updated_at) + VALUES (?,?,?,?,?,?)""", + [ + (account_id, sym, p["volume"], p.get("frozen", 0), + p["avg_price"], now) + for sym, p in positions.items() if p.get("volume", 0) > 0 + ], + ) + conn.commit() + + +def load_positions(db_path: str, account_id: int) -> list[dict]: + """API 读持仓快照 → [{symbol, volume, frozen, avg_price, updated_at}]。""" + with sqlite3.connect(db_path) as conn: + conn.row_factory = sqlite3.Row + cur = conn.execute( + "SELECT symbol, volume, frozen, avg_price, updated_at " + "FROM live_positions WHERE account_id=?", + (account_id,), + ) + return [dict(r) for r in cur.fetchall()] + + +# ----------------- live_balance ----------------- + +def save_balance( + db_path: str, account_id: int, date: str, cash: float, + market_value: float, total: float +) -> None: + with sqlite3.connect(db_path) as conn: + conn.execute( + """INSERT INTO live_balance + (account_id, date, cash, market_value, total) + VALUES (?,?,?,?,?)""", + (account_id, date, cash, market_value, total), + ) + conn.commit() + + +def list_balance(db_path: str, account_id: int) -> list[dict]: + with sqlite3.connect(db_path) as conn: + conn.row_factory = sqlite3.Row + cur = conn.execute( + "SELECT * FROM live_balance WHERE account_id=? ORDER BY date, id", + (account_id,), + ) + return [dict(r) for r in cur.fetchall()] + + +def get_last_balance(db_path: str, account_id: int) -> dict | None: + """最新一条账户快照(API /live/{aid}/account)。""" + with sqlite3.connect(db_path) as conn: + conn.row_factory = sqlite3.Row + cur = conn.execute( + "SELECT account_id, date, cash, market_value, total " + "FROM live_balance WHERE account_id=? ORDER BY id DESC LIMIT 1", + (account_id,), + ) + row = cur.fetchone() + return dict(row) if row else None + + +def get_first_balance(db_path: str, account_id: int) -> dict | None: + """最早一条账户快照(收益率基线;首快照 total = baseline)。""" + with sqlite3.connect(db_path) as conn: + conn.row_factory = sqlite3.Row + cur = conn.execute( + "SELECT account_id, date, cash, market_value, total " + "FROM live_balance WHERE account_id=? ORDER BY id ASC LIMIT 1", + (account_id,), + ) + row = cur.fetchone() + return dict(row) if row else None + + +__all__ = [ + "init_db", "save_account", "list_accounts", "get_account", + "update_account_status", "list_running_accounts", + "save_trade", "list_trades", + "save_positions", "load_positions", + "save_balance", "list_balance", "get_last_balance", "get_first_balance", +] diff --git a/sanguo_live/runner.py b/sanguo_live/runner.py new file mode 100644 index 0000000..d799dcb --- /dev/null +++ b/sanguo_live/runner.py @@ -0,0 +1,421 @@ +"""实盘交易入口:connect → add_strategy → subscribe → init_all → start_all → 常驻。 + +配置来源(优先级递增): + 1) ``sanguo_live/runner.py`` 内 DEFAULT_CONFIG + 2) ``config/live.yaml``(可选) + 3) 环境变量 ``SANGUO_QMT_ACCOUNT`` / ``SANGUO_QMT_PATH`` + +用法: + python -m sanguo_live # 用 config/live.yaml + python -m sanguo_live /path/to/cfg.yaml # 指定配置 + python -m sanguo_live --supervisor [db] # DB 驱动 supervisor + SANGUO_QMT_ACCOUNT=66639661 python -m sanguo_live + +注意:真实下单需要 miniQMT 同机运行且在交易时段。非交易时段运行只校验链路搭建。 +""" +from __future__ import annotations + +import logging +import os +import signal +import sys +import time +from datetime import datetime +from pathlib import Path +from typing import Any + +import yaml + +from sanguo_live.engine import LiveTradingEngine +from sanguo_live.strategies import AShareDoubleMaStrategy + +logger = logging.getLogger(__name__) + +DEFAULT_CONFIG_PATH = ( + Path(__file__).resolve().parent.parent / "config" / "live.yaml" +) + +DEFAULT_CONFIG: dict[str, Any] = { + "account": "", + "mini_path": "", + "strategy_name": "dm_15min_default", + "vt_symbol": "600000.SSE", + "strategy_class": "AShareDoubleMaStrategy", + # 连接后 sleep(秒):等 QmtGateway 完成 contract 拉取 + "connect_wait_sec": 10, + # init_all 后 sleep(秒):等策略 load_bar 加载历史 + "init_wait_sec": 60, + "setting": { + "fast_window": 10, + "slow_window": 20, + "window": 15, # BarGenerator 分钟窗口(A 股 15min) + "size": 100, # 1 手 = 100 股 + "forbid_short": True, + }, +} + +# 策略类注册表(可扩展) +_STRATEGY_REGISTRY: dict[str, type] = { + "AShareDoubleMaStrategy": AShareDoubleMaStrategy, +} + + +def load_config(path: str | Path | None = None) -> dict[str, Any]: + """加载配置:yaml 文件 + 环境变量覆盖(env 优先)。""" + cfg: dict[str, Any] = {k: (dict(v) if isinstance(v, dict) else v) + for k, v in DEFAULT_CONFIG.items()} + p = Path(path) if path else DEFAULT_CONFIG_PATH + if p.exists(): + try: + with open(p, encoding="utf-8") as f: + file_cfg = yaml.safe_load(f) or {} + except OSError as e: + logger.warning("读取配置失败 %s: %s(使用默认)", p, e) + file_cfg = {} + for k, v in file_cfg.items(): + if k == "setting" and isinstance(v, dict): + cfg["setting"].update(v) + else: + cfg[k] = v + # env 优先 + if os.environ.get("SANGUO_QMT_ACCOUNT"): + cfg["account"] = os.environ["SANGUO_QMT_ACCOUNT"] + if os.environ.get("SANGUO_QMT_PATH"): + cfg["mini_path"] = os.environ["SANGUO_QMT_PATH"] + return cfg + + +def build_strategy_class(name: str) -> type: + """策略类名 → 类。未知类抛 ValueError。""" + if name not in _STRATEGY_REGISTRY: + raise ValueError( + f"未知策略类: {name}; 可用: {list(_STRATEGY_REGISTRY)}" + ) + return _STRATEGY_REGISTRY[name] + + +def _install_signal_handlers(engine: LiveTradingEngine) -> None: + """SIGINT / SIGTERM → stop_all + close + exit。""" + + def _shutdown(signum: int, frame: Any) -> None: + logger.info("收到信号 %s,停止所有策略并退出", signum) + try: + engine.stop_all() + engine.close() + finally: + sys.exit(0) + + # SIGINT (Ctrl+C) 全平台;SIGTERM 仅 POSIX(Windows 上 Python 有定义但语义弱) + signal.signal(signal.SIGINT, _shutdown) + if hasattr(signal, "SIGTERM"): + try: + signal.signal(signal.SIGTERM, _shutdown) + except (ValueError, OSError): + pass # 非主线程或 Windows 子进程 — 忽略 + + +def run(config_path: str | Path | None = None) -> None: + """主流程:connect → add → subscribe → init → start → 常驻循环。""" + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + ) + cfg = load_config(config_path) + + if not cfg["account"]: + logger.error( + "未配置 QMT 交易账号。请设 SANGUO_QMT_ACCOUNT 或在 %s 写 account", + DEFAULT_CONFIG_PATH, + ) + sys.exit(2) + + engine = LiveTradingEngine() + + # 优雅退出 + _install_signal_handlers(engine) + + # 1) 连接 miniQMT + engine.connect({"交易账号": cfg["account"], "mini路径": cfg["mini_path"]}) + logger.info("等待 QMT 连接就绪 %ds...", cfg["connect_wait_sec"]) + time.sleep(int(cfg["connect_wait_sec"])) + + # 2) 注册策略 + strategy_cls = build_strategy_class(cfg["strategy_class"]) + engine.add_strategy( + strategy_cls, + cfg["strategy_name"], + cfg["vt_symbol"], + cfg["setting"], + ) + + # 3) 订阅行情(tick → BarGenerator → 15min bar → on_bar) + engine.subscribe([cfg["vt_symbol"]]) + + # 4) 初始化策略(load_bar 拉 10 天历史) + engine.init_all() + logger.info("等待策略 init 完成 %ds...", cfg["init_wait_sec"]) + time.sleep(int(cfg["init_wait_sec"])) + + # 5) 启动策略,进入实盘 + engine.start_all() + logger.info("=== 实盘已启动 (策略=%s 标的=%s)。Ctrl+C 退出 ===", + cfg["strategy_name"], cfg["vt_symbol"]) + + # 6) 常驻:主线程保活,行情/下单都在 EventEngine 工作线程 + while True: + time.sleep(10) + + +__all__ = ["run", "load_config", "build_strategy_class", + "DEFAULT_CONFIG", "DEFAULT_CONFIG_PATH", "run_supervisor", + "default_supervisor_db_path"] + + +# ============================================================================= +# supervisor 模式:DB 驱动,独立常驻进程轮询 live_accounts.status +# (task #4 持久化扩展)。API 只改 status 字段,supervisor 据此起停 engine。 +# ============================================================================= + +_SUPERVISOR_CONFIG_PATH = ( + Path(__file__).resolve().parent.parent / "config" / "data_platform.yaml" +) + + +def default_supervisor_db_path() -> str: + """从 config/data_platform.yaml 的 live_trading.db_path 读;fallback 主库。""" + try: + if _SUPERVISOR_CONFIG_PATH.exists(): + with open(_SUPERVISOR_CONFIG_PATH, encoding="utf-8") as f: + cfg = yaml.safe_load(f) or {} + lt = (cfg.get("live_trading") or {}) + if lt.get("db_path"): + return str(lt["db_path"]) + dp = cfg.get("data_paths") or {} + if dp.get("vnpy_db"): + return str(dp["vnpy_db"]) + except OSError as e: + logger.warning("读 supervisor 配置失败: %s", e) + return "quant_trading.db" + + +def _account_to_cfg(account_row: dict[str, Any]) -> dict[str, Any]: + """live_accounts 行 → runner 内部 cfg 结构(setting JSON 解析)。""" + import json as _json + try: + setting = _json.loads(account_row.get("setting") or "{}") + except (ValueError, TypeError): + setting = {} + return { + "account": account_row.get("account", ""), + "mini_path": account_row.get("mini_path", ""), + "strategy_name": account_row.get("strategy_name", ""), + "vt_symbol": account_row.get("vt_symbol", ""), + "strategy_class": account_row.get("strategy_class", + "AShareDoubleMaStrategy"), + "connect_wait_sec": int(account_row.get("connect_wait_sec", 10)), + "init_wait_sec": int(account_row.get("init_wait_sec", 60)), + "setting": setting, + } + + +def _start_engine_for_account(account_row: dict[str, Any]) -> LiveTradingEngine: + """根据 live_accounts 行起 LiveTradingEngine(connect→add→subscribe→init→start)。""" + from sanguo_live.engine import LiveTradingEngine + + cfg = _account_to_cfg(account_row) + engine = LiveTradingEngine() + engine.connect({"交易账号": cfg["account"], "mini路径": cfg["mini_path"]}) + logger.info("[supervisor] 等待 QMT 连接就绪 %ds (account=%s)...", + cfg["connect_wait_sec"], cfg["account"]) + time.sleep(cfg["connect_wait_sec"]) + + strategy_cls = build_strategy_class(cfg["strategy_class"]) + engine.add_strategy(strategy_cls, cfg["strategy_name"], + cfg["vt_symbol"], cfg["setting"]) + engine.subscribe([cfg["vt_symbol"]]) + engine.init_all() + logger.info("[supervisor] 等待策略 init %ds...", cfg["init_wait_sec"]) + time.sleep(cfg["init_wait_sec"]) + engine.start_all() + logger.info("[supervisor] engine 已启动 (account=%s strategy=%s)", + cfg["account"], cfg["strategy_name"]) + return engine + + +def _register_trade_handler( + engine: LiveTradingEngine, account_id: int, db_path: str +) -> Any: + """注册 EVENT_TRADE 回调:成交落 live_trades。返回 handler(供 unregister)。 + + EVENT_TRADE 定义在 ``vnpy.trader/event.py``(常量 "eTrade."), + ``vnpy.event`` 只导出 Event/EventEngine/EVENT_TIMER — 从 vnpy.event + import EVENT_TRADE 会 ImportError,导致成交回调静默不注册。 + """ + from sanguo_live.persistence import save_trade + + try: + from vnpy.trader.event import EVENT_TRADE # type: ignore + except ImportError: + logger.warning( + "[supervisor] 无 vnpy.trader.event,EVENT_TRADE 回调未注册" + ) + return None + + def _on_trade(event: Any) -> None: + try: + t = event.data + save_trade(db_path, account_id, { + "strategy_name": "", + "symbol": getattr(t, "vt_symbol", "") or "", + "direction": _enum_tail(getattr(t, "direction", "")), + "offset": _enum_tail(getattr(t, "offset", "")), + "price": float(getattr(t, "price", 0)), + "volume": float(getattr(t, "volume", 0)), + "traded_at": (t.datetime.isoformat() + if getattr(t, "datetime", None) else ""), + "vt_tradeid": getattr(t, "vt_tradeid", ""), + }) + except Exception as e: # noqa: BLE001 + logger.warning("[supervisor] save_trade 失败: %s", e) + + engine.event_engine.register(EVENT_TRADE, _on_trade) + return _on_trade + + +def _enum_tail(val: Any) -> str: + """Direction.LONG → 'long';Offset.OPEN → 'open';非 enum → str(val).lower()。""" + name = getattr(val, "name", None) + if name: + return str(name).lower() + return str(val).lower() if val else "" + + +def _snapshot_to_db( + engine: LiveTradingEngine, account_id: int, db_path: str +) -> None: + """定时把 OMS 持仓 + 账户快照落库(供 API 读)。""" + from sanguo_live.persistence import save_positions, save_balance + + try: + positions: dict[str, dict] = {} + for p in engine.get_positions(): + sym = getattr(p, "vt_symbol", "") or getattr(p, "symbol", "") + if not sym: + continue + # A 股只关心多头持仓(PositionDirection.LONG / NET) + direction = getattr(p, "direction", None) + dname = getattr(direction, "name", "") + if dname == "SHORT": + continue + vol = float(getattr(p, "volume", 0) or 0) + if vol <= 0: + continue + positions[sym] = { + "volume": vol, + "frozen": float(getattr(p, "frozen", 0) or 0), + "avg_price": float(getattr(p, "price", 0) or 0), + } + save_positions(db_path, account_id, positions) + + date_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + for acc in engine.get_all_accounts(): + total = float(getattr(acc, "balance", 0) or 0) + cash = float(getattr(acc, "available", 0) or 0) + save_balance(db_path, account_id, date_str, cash, + market_value=max(total - cash, 0.0), total=total) + except Exception as e: # noqa: BLE001 + logger.warning("[supervisor] snapshot 落库失败 (account=%s): %s", + account_id, e) + + +def _stop_engine(engine: LiveTradingEngine) -> None: + """stop_all + close(容错)。""" + try: + engine.stop_all() + finally: + try: + engine.close() + except Exception as e: # noqa: BLE001 + logger.warning("[supervisor] engine.close 异常: %s", e) + + +def run_supervisor( + db_path: str | None = None, + poll_interval_sec: float = 5.0, + snapshot_interval_sec: float = 30.0, +) -> None: + """DB 驱动的 supervisor 常驻进程。 + + 轮询 ``live_accounts.status``: + - 新 running → 起 LiveTradingEngine + 注册 EVENT_TRADE 回调 + - 变 stopped → 停 engine + close + 定时(snapshot_interval_sec)把 OMS 持仓/账户落 DB 供 API 读。 + 信号(SIGINT/SIGTERM)→ 停所有 engine 后退出。 + + MVP:每实例一个 engine,表结构支持多行(多实例同时跑只是内存多 engine)。 + """ + from sanguo_live.persistence import ( + init_db, list_running_accounts, get_account, + update_account_status, + ) + + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + ) + db = db_path or default_supervisor_db_path() + init_db(db) + logger.info("[supervisor] 启动 (db=%s poll=%.1fs snapshot=%.1fs)", + db, poll_interval_sec, snapshot_interval_sec) + + engines: dict[int, LiveTradingEngine] = {} + stop_flag = {"stop": False} + + def _shutdown(signum: int, frame: Any) -> None: + logger.info("[supervisor] 收到信号 %s,停止所有 engine", signum) + stop_flag["stop"] = True + + signal.signal(signal.SIGINT, _shutdown) + if hasattr(signal, "SIGTERM"): + try: + signal.signal(signal.SIGTERM, _shutdown) + except (ValueError, OSError): + pass + + last_snapshot: float = 0.0 + while not stop_flag["stop"]: + now = time.time() + # 1) 同步 status + running_ids = {r["id"] for r in list_running_accounts(db)} + # 启动新 running + for aid in running_ids - engines.keys(): + acc = get_account(db, aid) + if not acc: + continue + try: + eng = _start_engine_for_account(acc) + _register_trade_handler(eng, aid, db) + engines[aid] = eng + except Exception as e: # noqa: BLE001 + logger.error("[supervisor] 起 engine 失败 (account=%s): %s", + aid, e) + update_account_status(db, aid, "stopped", str(e)) + # 停止变 stopped 的 + for aid in list(engines.keys() - running_ids): + logger.info("[supervisor] 停止 engine (account=%s)", aid) + _stop_engine(engines.pop(aid)) + + # 2) 定时 snapshot + if now - last_snapshot >= snapshot_interval_sec: + for aid, eng in engines.items(): + _snapshot_to_db(eng, aid, db) + last_snapshot = now + + time.sleep(poll_interval_sec) + + # 退出清理 + for aid, eng in engines.items(): + logger.info("[supervisor] 退出清理 (account=%s)", aid) + _stop_engine(eng) + engines.clear() + logger.info("[supervisor] 已退出") diff --git a/sanguo_live/strategies.py b/sanguo_live/strategies.py new file mode 100644 index 0000000..0c0d941 --- /dev/null +++ b/sanguo_live/strategies.py @@ -0,0 +1,118 @@ +"""A 股 15min 双均线策略(AShareDoubleMaStrategy)。 + +为什么不直接用 ``vnpy_ctastrategy.strategies.DoubleMaStrategy``: +- 标准 DoubleMa 在 ``on_init`` 里 ``BarGenerator(self.on_bar)`` —— 无 ``window`` 参数, + 只合成 1min bar;且其 ``parameters = ["fast_window", "slow_window"]`` 不暴露周期配置, + **无法通过 setting 切到 15min**。 +- 本策略改用 ``BarGenerator(self._on_1min_bar, window=15, on_window_bar=self.on_bar)``, + 让 ``on_bar`` 直接收 15min bar,通过 ``window`` 参数(默认 15)可配。 +- 定寸 + 禁做空由 :class:`AShareCtaTemplate` 保证。 +- A 股不能做空,死叉时只平多,不开空(原 DoubleMa 的 ``pos<0 → cover+buy`` 分支删除)。 +""" +from __future__ import annotations + +import logging + +logger = logging.getLogger(__name__) + +try: + from vnpy.trader.utility import BarGenerator, ArrayManager # type: ignore + from vnpy.trader.constant import Interval # type: ignore + from vnpy.trader.object import BarData, TickData # type: ignore + _DEPS_OK = True +except ImportError: # 本机 dev 容错 + BarGenerator = None # type: ignore + ArrayManager = None # type: ignore + Interval = None # type: ignore + BarData = None # type: ignore + TickData = None # type: ignore + _DEPS_OK = False + +from sanguo_live.base_template import AShareCtaTemplate + + +class AShareDoubleMaStrategy(AShareCtaTemplate): + """A 股 双均线 15min 策略。 + + 金叉(fast 上穿 slow)且无持仓 → 买开 1 手; + 死叉(fast 下穿 slow)且持多 → 卖平 1 手; + A 股禁做空 → 不开空单(基类 short 已拦截)。 + """ + + author = "sanguo_live" + + fast_window: int = 10 + slow_window: int = 20 + # BarGenerator 分钟窗口。15 表示 15min。A 股支持 2/3/5/6/10/15/20/30(必须能整除 60)。 + window: int = 15 + + # 把基类的 size/forbid_short 一起暴露,update_setting 才会回填全部字段。 + parameters = ["fast_window", "slow_window", "window", "size", "forbid_short"] + variables = ["fast_ma0", "fast_ma1", "slow_ma0", "slow_ma1"] + + fast_ma0: float = 0.0 + fast_ma1: float = 0.0 + slow_ma0: float = 0.0 + slow_ma1: float = 0.0 + + def on_init(self) -> None: + self.write_log("AShareDoubleMa 策略初始化") + # 关键:用 BarGenerator 把 1min bar 合成 window 分钟 bar 后回调 self.on_bar。 + # _on_1min_bar 故意空实现 —— 我们不处理 1min,只让它喂养 BarGenerator。 + self.bg: BarGenerator = BarGenerator( + self._on_1min_bar, + window=self.window, + on_window_bar=self.on_bar, + ) + self.am: ArrayManager = ArrayManager() + self.load_bar(10) + + def on_start(self) -> None: + self.write_log("AShareDoubleMa 策略启动") + self.put_event() + + def on_stop(self) -> None: + self.write_log("AShareDoubleMa 策略停止") + self.put_event() + + def on_tick(self, tick: TickData) -> None: + """tick 推入 BarGenerator,由其合成 1min 与 window 分钟 bar。""" + self.bg.update_tick(tick) + + def _on_1min_bar(self, bar: BarData) -> None: + """1min bar 回调:故意忽略(由 BarGenerator 内部累积合成 15min)。""" + return + + def on_bar(self, bar: BarData) -> None: + """15min(或 setting 配置的 window)bar 回调。""" + self.cancel_all() + + am: ArrayManager = self.am + am.update_bar(bar) + if not am.inited: + return + + fast_ma = am.sma(self.fast_window, array=True) + self.fast_ma0 = fast_ma[-1] + self.fast_ma1 = fast_ma[-2] + + slow_ma = am.sma(self.slow_window, array=True) + self.slow_ma0 = slow_ma[-1] + self.slow_ma1 = slow_ma[-2] + + cross_over = self.fast_ma0 > self.slow_ma0 and self.fast_ma1 < self.slow_ma1 + cross_below = self.fast_ma0 < self.slow_ma0 and self.fast_ma1 > self.slow_ma1 + + if cross_over: + if self.pos == 0: + self.buy(bar.close_price, 1) + # A 股不可做空:原 DoubleMa 的 pos<0 分支(cover+buy)省略 + elif cross_below: + if self.pos > 0: + self.sell(bar.close_price, 1) + # A 股不可做空:原 DoubleMa 的 pos==0 short 分支省略 + + self.put_event() + + +__all__ = ["AShareDoubleMaStrategy"] diff --git a/sanguo_portfolio/runner_backtest.py b/sanguo_portfolio/runner_backtest.py index 9d1ac9d..a2cf111 100644 --- a/sanguo_portfolio/runner_backtest.py +++ b/sanguo_portfolio/runner_backtest.py @@ -5,6 +5,10 @@ python -m sanguo_portfolio.runner_backtest \\ --start 2020-01-01 --end 2024-12-31 --cash 1000000 +JSON 输出(供 SSH 捕获,前端 MVP 用): + python -m sanguo_portfolio.runner_backtest --json \\ + --start 2024-01-01 --end 2024-02-29 --cash 1000000 + Mac 没装 xtquant,这里仅作为入口脚本(测试用 mock,实际跑 rsync 到 VPS)。 """ from __future__ import annotations @@ -14,6 +18,7 @@ import os os.environ.setdefault("DEFAULT_DATA_PROVIDER", "miniqmt") import argparse +import json import logging from typing import Any, Dict @@ -35,6 +40,10 @@ def parse_args() -> argparse.Namespace: "--result-file", default="docs/portfolio_backtest_result.md", help="结果输出文件(.md)", ) + p.add_argument( + "--json", action="store_true", + help="JSON 模式:print(json.dumps(result)) 到 stdout,供 SSH 捕获", + ) return p.parse_args() @@ -91,7 +100,7 @@ def run_backtest(args: argparse.Namespace) -> Dict[str, Any]: BulletTrade 的 BacktestEngine 接受 strategy_file 或 initialize 等函数。 我们把 AllWeatherStrategy 包成 initialize 函数:initialize 闭包挂 run_daily 等。 """ - from bullet_trade.core import BacktestEngine # type: ignore + from bullet_trade import BacktestEngine # type: ignore from bullet_trade.data.api import set_data_provider # type: ignore from .strategies import AllWeatherStrategy @@ -145,8 +154,9 @@ def run_backtest(args: argparse.Namespace) -> Dict[str, Any]: ) result = engine.run() - # 输出结果摘要到 markdown - _write_result_md(result, args.result_file, args) + # 输出结果摘要到 markdown(JSON 模式时 result_file="" 跳过) + if getattr(args, "result_file", ""): + _write_result_md(result, args.result_file, args) return result @@ -181,10 +191,193 @@ def _write_result_md(result: Dict[str, Any], path: str, args: argparse.Namespace logger.warning("写结果文件失败: %s", exc) +def run_backtest_json(params: Dict[str, Any]) -> Dict[str, Any]: + """JSON 入口(供 SSH 触发,前端 MVP 用)。 + + Args: + params: { + pool: 标的池(暂未实际使用,占位), + start_date, end_date: YYYY-MM-DD, + initial_cash: 初始资金, + } + + Returns: + { + "strategy": "all_weather", + "period": {"start": ..., "end": ..., "trading_days": N}, + "stocks_selected": [{"code":..., "name":...}, ...], # 末日持仓 + "trades": [{date, code, side, amount, price, ...}, ...], + "equity_curve": [{"date":..., "equity":...}, ...], + "metrics": {total_return, annual_return, max_drawdown, sharpe, ...}, + } + """ + # 构造一个 Namespace 复用 run_backtest + args = argparse.Namespace( + start=params.get("start_date", "2024-01-01"), + end=params.get("end_date", "2024-02-29"), + cash=float(params.get("initial_cash", 1_000_000.0)), + benchmark=params.get("benchmark", "000300.XSHG"), + frequency="day", + provider_config="{}", + result_file="", # JSON 模式不写 md + ) + raw = run_backtest(args) + + summary = raw.get("summary", {}) if isinstance(raw, dict) else {} + metrics = _extract_metrics(summary) + + # 净值曲线:daily_records 是 DataFrame,index=date,列含 total_value + equity_curve = _extract_equity_curve(raw.get("daily_records")) + + # 选股(末日持仓):daily_positions 最后一日 + stocks_selected = _extract_last_positions(raw.get("daily_positions")) + + # 成交明细 + trades = _extract_trades(raw.get("trades")) + + meta = raw.get("meta", {}) if isinstance(raw, dict) else {} + return { + "strategy": "all_weather", + "period": { + "start": meta.get("start_date", args.start), + "end": meta.get("end_date", args.end), + "trading_days": len(equity_curve), + }, + "stocks_selected": stocks_selected, + "trades": trades, + "equity_curve": equity_curve, + "metrics": metrics, + "raw_summary": summary, + } + + +def _to_float(v: Any) -> float | None: + """从 string/number 提取 float,失败返 None。bullet-trade summary 多为 '12.34%' 字符串。""" + if v is None: + return None + if isinstance(v, (int, float)): + return float(v) + s = str(v).strip().replace("%", "").replace(",", "") + try: + return float(s) + except (TypeError, ValueError): + return None + + +def _extract_metrics(summary: Dict[str, Any]) -> Dict[str, float | None]: + """bullet-trade summary 用中文 key('策略收益'/'最大回撤'/...)。 + 百分比按字面数值(12.34% → 12.34),前端按需 /100 显示。 + """ + return { + "total_return": _to_float(summary.get("策略收益")), + "annual_return": _to_float(summary.get("策略年化收益")), + "max_drawdown": _to_float(summary.get("最大回撤")), + "sharpe": _to_float(summary.get("夏普比率")), + "win_rate_daily": _to_float(summary.get("日胜率")), + "win_rate_trade": _to_float(summary.get("交易胜率")), + "trading_days": _to_float(summary.get("交易天数")), + } + + +def _extract_equity_curve(daily_records: Any) -> list[Dict[str, Any]]: + """daily_records: DataFrame,index=date,列含 total_value。""" + out: list[Dict[str, Any]] = [] + if daily_records is None: + return out + try: + import pandas as pd # type: ignore + if isinstance(daily_records, pd.DataFrame): + df = daily_records.reset_index() + date_col = "date" if "date" in df.columns else df.columns[0] + val_col = "total_value" if "total_value" in df.columns else None + if val_col is None: + return out + for _, row in df.iterrows(): + d = row[date_col] + out.append({ + "date": getattr(d, "strftime", lambda f: str(d))("%Y-%m-%d"), + "equity": float(row[val_col]), + }) + except Exception as exc: + logger.warning("解析 equity_curve 失败: %s", exc) + return out + + +def _extract_last_positions(daily_positions: Any) -> list[Dict[str, Any]]: + """daily_positions: DataFrame,列含 date/code/amount/avg_cost/price/value。 + 取最后一日的非零持仓作为选股名单。""" + out: list[Dict[str, Any]] = [] + if daily_positions is None: + return out + try: + import pandas as pd # type: ignore + if isinstance(daily_positions, pd.DataFrame) and not daily_positions.empty: + df = daily_positions + if "date" in df.columns: + last_date = df["date"].max() + df = df[df["date"] == last_date] + for _, row in df.iterrows(): + amt = row.get("amount", 0) + if amt is None or float(amt) <= 0: + continue + out.append({ + "code": str(row.get("code", "")), + "name": str(row.get("code", "")), # name 字段 bullet-trade 没存,前端展示 code + "amount": float(amt), + "avg_cost": float(row.get("avg_cost", 0) or 0), + "price": float(row.get("price", 0) or 0), + "value": float(row.get("value", 0) or 0), + }) + except Exception as exc: + logger.warning("解析 last_positions 失败: %s", exc) + return out + + +def _extract_trades(trades: Any) -> list[Dict[str, Any]]: + """trades: list[Trade],用 __dict__ 或属性兜底提取关键字段。""" + out: list[Dict[str, Any]] = [] + if not trades: + return out + keys = ("datetime", "date", "code", "side", "action", "amount", + "filled_amount", "price", "filled_price", "commission", "status") + for t in trades: + item: Dict[str, Any] = {} + for k in keys: + v = None + if hasattr(t, k): + v = getattr(t, k) + elif isinstance(t, dict): + v = t.get(k) + if v is None: + continue + # datetime 类转字符串 + if hasattr(v, "strftime"): + v = v.strftime("%Y-%m-%d %H:%M:%S") + try: + if isinstance(v, (int, float)): + v = float(v) + except Exception: + pass + item[k] = v + if item: + out.append(item) + return out + + def main() -> None: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") args = parse_args() - run_backtest(args) + if args.json: + # JSON 模式:stderr 仍打日志,stdout 只输出 JSON(供 SSH 捕获) + result = run_backtest_json({ + "start_date": args.start, + "end_date": args.end, + "initial_cash": args.cash, + "benchmark": args.benchmark, + }) + print(json.dumps(result, ensure_ascii=False, default=str)) + else: + run_backtest(args) if __name__ == "__main__": diff --git a/tests/test_live_api.py b/tests/test_live_api.py new file mode 100644 index 0000000..2e69dc2 --- /dev/null +++ b/tests/test_live_api.py @@ -0,0 +1,370 @@ +"""实盘模拟 API + 持久化单测（task #4）。 + +两层： + (1) persistence CRUD —— sqlite tmp，纯 Python，Mac 跑通； + (2) routes_live API —— FastAPI TestClient，不实例化 LiveTradingEngine + （supervisor 才起 engine，本模块只测 DB CRUD 路由）。 + +Mac 跑：``pytest tests/test_live_api.py -v`` +""" +from __future__ import annotations + +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 + + +# ============================================================================= +# 层 1：persistence CRUD +# ============================================================================= + +def _db(tmp_path) -> str: + from sanguo_live.persistence import init_db + db = os.path.join(str(tmp_path), "live.db") + init_db(db) + return db + + +def test_save_and_get_account(tmp_path): + from sanguo_live.persistence import save_account, get_account + + db = _db(tmp_path) + aid = save_account(db, { + "name": "live1", "account": "12345678", + "vt_symbol": "600000.SSE", "strategy_name": "dm1", + "setting": {"fast_window": 5}, "initial_capital": 5e5, + }) + assert aid > 0 + acc = get_account(db, aid) + assert acc["account"] == "12345678" + assert acc["status"] == "stopped" # 默认 stopped + assert acc["vt_symbol"] == "600000.SSE" + assert "\"fast_window\": 5" in acc["setting"] # JSON 字符串 + + +def test_list_accounts_and_default_status(tmp_path): + from sanguo_live.persistence import save_account, list_accounts + + db = _db(tmp_path) + save_account(db, {"account": "1", "strategy_name": "s1"}) + save_account(db, {"account": "2", "strategy_name": "s2"}) + rows = list_accounts(db) + assert len(rows) == 2 + # DESC 排序：最新建的在前 + assert rows[0]["account"] == "2" + assert all(r["status"] == "stopped" for r in rows) + + +def test_update_account_status(tmp_path): + from sanguo_live.persistence import ( + save_account, update_account_status, get_account, + list_running_accounts, + ) + + db = _db(tmp_path) + aid = save_account(db, {"account": "999", "strategy_name": "s"}) + update_account_status(db, aid, "running") + assert get_account(db, aid)["status"] == "running" + assert len(list_running_accounts(db)) == 1 + update_account_status(db, aid, "stopped", "test error") + assert get_account(db, aid)["status"] == "stopped" + assert get_account(db, aid)["error_msg"] == "test error" + assert list_running_accounts(db) == [] + + +def test_save_and_list_trades(tmp_path): + from sanguo_live.persistence import save_account, save_trade, list_trades + + db = _db(tmp_path) + aid = save_account(db, {"account": "1", "strategy_name": "s"}) + tid1 = save_trade(db, aid, {"symbol": "600000.SSE", "direction": "long", + "offset": "open", "price": 10.5, "volume": 100, + "traded_at": "2026-07-17T10:00:00", + "vt_tradeid": "T1"}) + tid2 = save_trade(db, aid, {"symbol": "600000.SSE", "direction": "short", + "offset": "close", "price": 11.0, "volume": 100, + "traded_at": "2026-07-17T11:00:00", + "vt_tradeid": "T2"}) + assert tid1 > 0 and tid2 > tid1 + trades = list_trades(db, aid) + assert len(trades) == 2 + assert trades[0]["vt_tradeid"] == "T1" + assert trades[1]["price"] == 11.0 + + +def test_save_positions_overwrites_snapshot(tmp_path): + """positions 覆盖式快照：第二次 save 完全替换第一次。""" + from sanguo_live.persistence import ( + save_account, save_positions, load_positions, + ) + + db = _db(tmp_path) + aid = save_account(db, {"account": "1", "strategy_name": "s"}) + save_positions(db, aid, { + "600000.SSE": {"volume": 100, "frozen": 0, "avg_price": 10.0}, + "000001.SZSE": {"volume": 200, "frozen": 50, "avg_price": 15.0}, + }) + pos = load_positions(db, aid) + assert len(pos) == 2 + # 覆盖（600000 减仓，000001 清仓） + save_positions(db, aid, { + "600000.SSE": {"volume": 50, "frozen": 0, "avg_price": 10.0}, + }) + pos2 = load_positions(db, aid) + assert len(pos2) == 1 + assert pos2[0]["symbol"] == "600000.SSE" + assert pos2[0]["volume"] == 50 + + +def test_save_positions_skips_zero_volume(tmp_path): + from sanguo_live.persistence import save_account, save_positions, load_positions + + db = _db(tmp_path) + aid = save_account(db, {"account": "1", "strategy_name": "s"}) + save_positions(db, aid, { + "600000.SSE": {"volume": 0, "frozen": 0, "avg_price": 0}, + "000001.SZSE": {"volume": 100, "frozen": 0, "avg_price": 15.0}, + }) + pos = load_positions(db, aid) + assert len(pos) == 1 + assert pos[0]["symbol"] == "000001.SZSE" + + +def test_save_and_get_last_balance(tmp_path): + from sanguo_live.persistence import ( + save_account, save_balance, list_balance, get_last_balance, + ) + + db = _db(tmp_path) + aid = save_account(db, {"account": "1", "strategy_name": "s"}) + save_balance(db, aid, "2026-07-17 10:00:00", 5e5, 1e5, 6e5) + save_balance(db, aid, "2026-07-17 11:00:00", 4e5, 2e5, 6e5) + all_bal = list_balance(db, aid) + assert len(all_bal) == 2 + last = get_last_balance(db, aid) + assert last["cash"] == 4e5 + assert last["total"] == 6e5 + assert last["date"] == "2026-07-17 11:00:00" + + +def test_get_last_balance_empty(tmp_path): + from sanguo_live.persistence import save_account, get_last_balance + + db = _db(tmp_path) + aid = save_account(db, {"account": "1", "strategy_name": "s"}) + assert get_last_balance(db, aid) is None + + +# ============================================================================= +# 层 2：routes_live API（TestClient，不依赖 vnpy） +# ============================================================================= + +def _client(tmp_path): + set_jwt_config(secret="t", expire_minutes=60) + db = os.path.join(str(tmp_path), "live_api.db") + app = create_app(db_path=db) + set_db_path(db) + return TestClient(app), create_token("admin") + + +def _auth(token): + return {"Authorization": f"Bearer {token}"} + + +def test_create_live(tmp_path): + c, token = _client(tmp_path) + resp = c.post( + "/api/v1/live/create", + json={ + "name": "live1", "account": "12345678", + "vt_symbol": "600000.SSE", "strategy_name": "dm1", + "setting": {"fast_window": 5}, "initial_capital": 5e5, + }, + headers=_auth(token), + ) + assert resp.status_code == 200 + body = resp.json() + assert body["account_id"] > 0 + assert body["status"] == "stopped" # create 后默认 stopped + + +def test_list_and_get_live(tmp_path): + c, token = _client(tmp_path) + aid = c.post( + "/api/v1/live/create", + json={"account": "123", "strategy_name": "dm1"}, + headers=_auth(token), + ).json()["account_id"] + # 列表 + lst = c.get("/api/v1/live", headers=_auth(token)).json() + assert len(lst["accounts"]) == 1 + assert lst["accounts"][0]["id"] == aid + assert lst["accounts"][0]["total_return"] is None # 无 balance + assert lst["accounts"][0]["position_count"] == 0 + # 详情 + detail = c.get(f"/api/v1/live/{aid}", headers=_auth(token)).json() + assert detail["account"] == "123" + assert detail["status"] == "stopped" + + +def test_start_and_stop(tmp_path): + c, token = _client(tmp_path) + aid = c.post( + "/api/v1/live/create", + json={"account": "123", "strategy_name": "dm1"}, + headers=_auth(token), + ).json()["account_id"] + # start + r = c.post(f"/api/v1/live/{aid}/start", headers=_auth(token)) + assert r.status_code == 200 + assert r.json()["status"] == "running" + assert c.get(f"/api/v1/live/{aid}/status", + headers=_auth(token)).json()["status"] == "running" + # stop + r = c.post(f"/api/v1/live/{aid}/stop", headers=_auth(token)) + assert r.json()["status"] == "stopped" + assert c.get(f"/api/v1/live/{aid}/status", + headers=_auth(token)).json()["status"] == "stopped" + + +def test_start_empty_account_rejected(tmp_path): + """account 字段空 → start 返回 400。""" + c, token = _client(tmp_path) + aid = c.post( + "/api/v1/live/create", + json={"account": "", "strategy_name": "dm1"}, + headers=_auth(token), + ).json()["account_id"] + r = c.post(f"/api/v1/live/{aid}/start", headers=_auth(token)) + assert r.status_code == 400 + + +def test_empty_trades_positions_account(tmp_path): + """新建实例：trades / positions / account 应返回空结构。""" + c, token = _client(tmp_path) + aid = c.post( + "/api/v1/live/create", + json={"account": "123", "strategy_name": "dm1"}, + headers=_auth(token), + ).json()["account_id"] + assert c.get(f"/api/v1/live/{aid}/trades", + headers=_auth(token)).json() == [] + assert c.get(f"/api/v1/live/{aid}/positions", + headers=_auth(token)).json() == [] + assert c.get(f"/api/v1/live/{aid}/account", + headers=_auth(token)).json() == {} + + +def test_get_live_404(tmp_path): + c, token = _client(tmp_path) + assert c.get("/api/v1/live/999", headers=_auth(token)).status_code == 404 + assert c.get("/api/v1/live/999/status", + headers=_auth(token)).status_code == 404 + assert c.post("/api/v1/live/999/start", + headers=_auth(token)).status_code == 404 + + +def test_unauthorized_401(tmp_path): + c, _ = _client(tmp_path) + assert c.get("/api/v1/live").status_code == 401 + assert c.post("/api/v1/live/create", + json={"account": "1", "strategy_name": "s"}).status_code == 401 + + +def test_create_live_mini_path_default_when_empty(monkeypatch, tmp_path): + """create 不传 mini_path → 后端 env/内置默认兜底,落库 mini_path 非空。 + + 避免空 mini_path 导致 connect=-1(task #6a 冒烟发现)。 + """ + monkeypatch.delenv("SANGUO_QMT_PATH", raising=False) + c, token = _client(tmp_path) + aid = c.post( + "/api/v1/live/create", + json={"account": "123", "strategy_name": "dm1"}, # 不传 mini_path + headers=_auth(token), + ).json()["account_id"] + from sanguo_live.persistence import get_account + db = os.path.join(str(tmp_path), "live_api.db") + acc = get_account(db, aid) + assert acc["mini_path"] # 非空 + assert "userdata_mini" in acc["mini_path"] # 内置默认 + + +def test_create_live_mini_path_env_fallback(monkeypatch, tmp_path): + """req.mini_path 空 → env SANGUO_QMT_PATH 兜底(优先于内置默认)。""" + monkeypatch.setenv("SANGUO_QMT_PATH", "/from/env/mini") + c, token = _client(tmp_path) + aid = c.post( + "/api/v1/live/create", + json={"account": "123", "strategy_name": "dm1", "mini_path": ""}, + headers=_auth(token), + ).json()["account_id"] + from sanguo_live.persistence import get_account + db = os.path.join(str(tmp_path), "live_api.db") + assert get_account(db, aid)["mini_path"] == "/from/env/mini" + + +def test_list_lives_total_return_uses_first_snapshot_baseline(tmp_path): + """list 收益率按首快照 total 为 baseline,不是 initial_capital。 + + 场景:initial_capital=6e5,但首快照 total=5e5(模拟入金后立刻记录)。 + 两条 balance:5e5 → 5.5e5,收益率应为 (5.5e5 - 5e5) / 5e5 = 0.1, + 而非按 initial_capital 6e5 算的 -0.0833。 + """ + from sanguo_live.persistence import save_balance + c, token = _client(tmp_path) + aid = c.post( + "/api/v1/live/create", + json={"account": "123", "strategy_name": "dm1", + "initial_capital": 6e5}, + headers=_auth(token), + ).json()["account_id"] + db = os.path.join(str(tmp_path), "live_api.db") + save_balance(db, aid, "2026-07-17 09:30:00", 5e5, 0, 5e5) # baseline + save_balance(db, aid, "2026-07-17 15:00:00", 5e5, 0.5e5, 5.5e5) + + item = c.get("/api/v1/live", headers=_auth(token)).json()["accounts"][0] + assert item["latest_equity"] == 5.5e5 + # (5.5e5 - 5e5) / 5e5 = 0.1 + assert abs(item["total_return"] - 0.1) < 1e-9 + + +def test_routes_reflect_db_writes(tmp_path): + """直接写 DB（模拟 supervisor 落库）→ API 路由读到。""" + from sanguo_live.persistence import ( + save_trade, save_positions, save_balance, + ) + + c, token = _client(tmp_path) + aid = c.post( + "/api/v1/live/create", + json={"account": "123", "strategy_name": "dm1", + "initial_capital": 6e5}, + headers=_auth(token), + ).json()["account_id"] + db = os.path.join(str(tmp_path), "live_api.db") + save_trade(db, aid, {"symbol": "600000.SSE", "direction": "long", + "offset": "open", "price": 10.0, "volume": 100, + "traded_at": "2026-07-17T10:00:00"}) + save_positions(db, aid, {"600000.SSE": {"volume": 100, "frozen": 0, + "avg_price": 10.0}}) + save_balance(db, aid, "2026-07-17 10:00:00", 5e5, 1e5, 6e5) + + trades = c.get(f"/api/v1/live/{aid}/trades", headers=_auth(token)).json() + assert len(trades) == 1 + assert trades[0]["price"] == 10.0 + pos = c.get(f"/api/v1/live/{aid}/positions", headers=_auth(token)).json() + assert len(pos) == 1 and pos[0]["symbol"] == "600000.SSE" + acc = c.get(f"/api/v1/live/{aid}/account", headers=_auth(token)).json() + assert acc["total"] == 6e5 + + # 列表汇总：有 balance 后 total_return 应非 None + lst = c.get("/api/v1/live", headers=_auth(token)).json() + item = lst["accounts"][0] + assert item["latest_equity"] == 6e5 + assert item["total_return"] == 0.0 # 6e5 == 初始 6e5 + assert item["position_count"] == 1 diff --git a/tests/test_live_engine.py b/tests/test_live_engine.py new file mode 100644 index 0000000..80133cd --- /dev/null +++ b/tests/test_live_engine.py @@ -0,0 +1,252 @@ +"""sanguo_live 单元测试。 + +分两层: + (1) 纯 Python 逻辑层 —— Mac dev 机也跑(config 解析、注册表、默认参数); + (2) 依赖 vnpy_ctastrategy 层 —— Mac 未装时单测级 skip,VPS 装齐则跑通。 + +Mac 跑:``pytest tests/test_live_engine.py -v``(层 1 全 pass + 层 2 skipped,exit 0)。 +VPS 跑:全部 pass(含定寸/禁做空/引擎装配)。 +""" +from __future__ import annotations + +import importlib + +import pytest + + +def _has_vnpy_cta() -> bool: + """Mac dev 机没装 vnpy_ctastrategy(只装在 VPS)。""" + try: + importlib.import_module("vnpy_ctastrategy") + return True + except ImportError: + return False + + +# 单测级 skip marker(模块级 importorskip 会跳过整个文件,误伤层 1) +needs_vnpy_cta = pytest.mark.skipif( + not _has_vnpy_cta(), + reason="本机未装 vnpy_ctastrategy(仅 VPS 有)— 跳过依赖它的单测", +) + + +# ============================================================================= +# 层 1:纯 Python 逻辑(Mac dev 机也跑) +# ============================================================================= + + +def test_module_import_tolerant(): + """``sanguo_live`` 包 import 不应崩(即便本机没 vnpy_ctastrategy)。""" + importlib.import_module("sanguo_live") + importlib.import_module("sanguo_live.base_template") + importlib.import_module("sanguo_live.runner") + # engine / strategies import 了 vnpy_qmt/vnpy_ctastrategy 的类绑定, + # 但都用 try/except 容错,模块本身能 import。 + importlib.import_module("sanguo_live.engine") + importlib.import_module("sanguo_live.strategies") + + +def test_default_config_fields(): + from sanguo_live.runner import DEFAULT_CONFIG + assert DEFAULT_CONFIG["strategy_class"] == "AShareDoubleMaStrategy" + assert DEFAULT_CONFIG["vt_symbol"] == "600000.SSE" + s = DEFAULT_CONFIG["setting"] + assert s["window"] == 15 + assert s["size"] == 100 + assert s["forbid_short"] is True + assert s["fast_window"] == 10 + assert s["slow_window"] == 20 + + +def test_load_config_env_override(monkeypatch): + """env SANGUO_QMT_ACCOUNT / SANGUO_QMT_PATH 优先于 yaml / 默认。""" + monkeypatch.setenv("SANGUO_QMT_ACCOUNT", "12345678") + monkeypatch.setenv("SANGUO_QMT_PATH", "/tmp/fake_mini") + from sanguo_live.runner import load_config + cfg = load_config("/nonexistent/path.yaml") # 文件不存在 → 走默认 + assert cfg["account"] == "12345678" + assert cfg["mini_path"] == "/tmp/fake_mini" + + +def test_load_config_yaml_merge(tmp_path): + """yaml 能覆盖默认 fast_window 等。""" + yaml_file = tmp_path / "live.yaml" + yaml_file.write_text( + "account: '99999999'\n" + "vt_symbol: '000001.SZSE'\n" + "setting:\n" + " fast_window: 5\n" + " slow_window: 30\n", + encoding="utf-8", + ) + from sanguo_live.runner import load_config + cfg = load_config(str(yaml_file)) + assert cfg["account"] == "99999999" + assert cfg["vt_symbol"] == "000001.SZSE" + assert cfg["setting"]["fast_window"] == 5 + assert cfg["setting"]["slow_window"] == 30 + # 未覆盖的字段保留默认 + assert cfg["setting"]["window"] == 15 + assert cfg["strategy_class"] == "AShareDoubleMaStrategy" + + +def test_build_strategy_class_known(): + from sanguo_live.runner import build_strategy_class + cls = build_strategy_class("AShareDoubleMaStrategy") + assert cls.__name__ == "AShareDoubleMaStrategy" + + +def test_build_strategy_class_unknown_raises(): + from sanguo_live.runner import build_strategy_class + with pytest.raises(ValueError, match="未知策略类"): + build_strategy_class("NoSuchStrategy_xyz") + + +def test_strategy_class_has_parameters(): + """AShareDoubleMaStrategy.parameters 必须暴露 size/forbid_short/window + + fast/slow_window(缺一个都会让 update_setting 漏字段)。""" + from sanguo_live.strategies import AShareDoubleMaStrategy + params = AShareDoubleMaStrategy.parameters + for required in ("fast_window", "slow_window", "window", + "size", "forbid_short"): + assert required in params, f"缺少 parameter: {required}" + + +# ============================================================================= +# 层 2:依赖 vnpy_ctastrategy(Mac skip,VPS 跑) +# ============================================================================= + + +class _FakeCtaEngine: + """记录 send_order 调用,模拟 CtaTemplate 依赖的 cta_engine。""" + + def __init__(self) -> None: + self.calls: list[tuple] = [] + + def send_order(self, strategy, direction, offset, price, volume, + stop=False, lock=False, net=False): + self.calls.append((direction, offset, price, volume, stop, lock, net)) + return [] + + def cancel_all(self, strategy): + return None + + +def _make_strategy(cls, setting=None): + """构造一个策略实例(trading=True,可发单)。 + + ``cls`` 必须是具体类(CtaTemplate 是 ABC,带抽象 on_init,不能直接实例化)。 + 用 ``_ConcreteAShare`` 包装 AShareCtaTemplate 来测基类定寸/禁做空逻辑。 + """ + strat = cls(_FakeCtaEngine(), "test_strat", "600000.SSE", setting or {}) + strat.trading = True + return strat + + +def _concrete_asare(): + """返回 AShareCtaTemplate 的一个具体子类(stub on_init/on_tick/on_bar)。""" + from sanguo_live.base_template import AShareCtaTemplate + + class _Concrete(AShareCtaTemplate): + author = "test" + + def on_init(self) -> None: # type: ignore[override] + return + + def on_tick(self, tick) -> None: # type: ignore[override] + return + + def on_bar(self, bar) -> None: # type: ignore[override] + return + + return _Concrete + + +@needs_vnpy_cta +def test_buy_volume_multiplied_by_size(): + """buy(1) 实际下单 volume=100(1 手 × size)。""" + strat = _make_strategy(_concrete_asare(), {"size": 100}) + strat.buy(10.0, 1) + assert len(strat.cta_engine.calls) == 1 + _, _, price, volume, *_ = strat.cta_engine.calls[0] + assert price == 10.0 + assert volume == 100 + + +@needs_vnpy_cta +def test_buy_custom_size_multiplier(): + """size=200 → buy(2) 下 400。""" + strat = _make_strategy(_concrete_asare(), {"size": 200}) + strat.buy(8.8, 2) + assert strat.cta_engine.calls[0][3] == 400 + + +@needs_vnpy_cta +def test_sell_volume_multiplied_by_size(): + """sell(平多)同样定寸。""" + strat = _make_strategy(_concrete_asare(), {"size": 100}) + strat.sell(11.0, 1) + assert strat.cta_engine.calls[0][3] == 100 + + +@needs_vnpy_cta +def test_cover_volume_multiplied_by_size(): + """cover 也定寸(策略逻辑误调时不至于下零股)。""" + strat = _make_strategy(_concrete_asare(), {"size": 100}) + strat.cover(11.0, 1) + assert strat.cta_engine.calls[0][3] == 100 + + +@needs_vnpy_cta +def test_short_blocked_by_default(): + """forbid_short=True(默认) → short 返回 [],不触达 send_order。""" + class _ExplodingEngine: + def send_order(self, *a, **kw): + raise AssertionError("short 不应到达 send_order") + + def write_log(self, msg, strategy=None): + return + + strat = _concrete_asare()(_ExplodingEngine(), "t", "600000.SSE", {}) + strat.trading = True + result = strat.short(10.0, 1) + assert result == [] + + +@needs_vnpy_cta +def test_short_passes_when_forbid_disabled(): + """forbid_short=False → short 定寸后透传到基类(A 股不推荐,仅测试逻辑)。""" + strat = _make_strategy(_concrete_asare(), + {"size": 100, "forbid_short": False}) + strat.short(10.0, 1) + assert len(strat.cta_engine.calls) == 1 + assert strat.cta_engine.calls[0][3] == 100 # 1 手 × 100 + + +@needs_vnpy_cta +def test_double_ma_strategy_uses_15min_window(): + """AShareDoubleMaStrategy 默认 window=15(不是 1min)。""" + from sanguo_live.strategies import AShareDoubleMaStrategy + strat = _make_strategy(AShareDoubleMaStrategy, {}) + assert strat.window == 15 + assert strat.size == 100 + assert strat.forbid_short is True + + +@needs_vnpy_cta +def test_engine_assembly_and_close(): + """LiveTradingEngine 初始化 → MainEngine 装入 QMT gateway + CTA app,close 干净退出。 + + 会真启动 EventEngine 线程,测试结束必须 close。 + """ + from sanguo_live.engine import LiveTradingEngine + eng = LiveTradingEngine() + try: + assert eng.cta_engine is not None + assert "QMT" in eng.main_engine.gateways + # 查询方法不抛(连接前可能返回空) + assert isinstance(eng.get_all_accounts(), list) + assert isinstance(eng.get_positions(), list) + assert isinstance(eng.get_orders(), list) + finally: + eng.close()