diff --git a/backend/app/api/monitor_rules.py b/backend/app/api/monitor_rules.py index ce32077a..a362cccc 100644 --- a/backend/app/api/monitor_rules.py +++ b/backend/app/api/monitor_rules.py @@ -56,6 +56,11 @@ def _sync_engine(request: Request) -> None: for r in monitor_rules.load_all(_data_dir(request)) ] engine.set_rules(rules) + # set_rules 全量覆盖会清掉模拟盘合成规则 → 每次必须重注入; + # 覆盖集变化 (新建/删除 strategy 规则) 也会改变去重结果。 + from app.strategy import paper_watch + + paper_watch.sync_from_state(request.app.state, _data_dir(request)) # ── Pydantic 模型 ─────────────────────────────────────── diff --git a/backend/app/api/paper.py b/backend/app/api/paper.py index 13e46b1d..147641c0 100644 --- a/backend/app/api/paper.py +++ b/backend/app/api/paper.py @@ -126,6 +126,9 @@ def delete_account(request: Request, force: bool = False, account: str = Query(p result, err = paper.delete_account(_data_dir(request), _acc(request, account), force=force) if err: raise HTTPException(status_code=400 if "force" in (err or "") else 404, detail=err) + from app.strategy import paper_watch + + paper_watch.sync_from_state(request.app.state, _data_dir(request)) # 账户删了, 绑定集变了 return {"ok": True, "deleted": _acc(request, account), **(result or {})} @@ -369,28 +372,31 @@ def list_auto_rules(request: Request, account: str = Query(paper.DEFAULT_ACCOUNT @router.post("/auto_rules") def create_auto_rule(request: Request, body: AutoRuleModel, account: str = Query(paper.DEFAULT_ACCOUNT_ID)): - from app.strategy import paper_auto + from app.strategy import paper_auto, paper_watch try: rule = paper_auto.create_auto_rule(_data_dir(request), body.model_dump(), _acc(request, account)) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) from e + paper_watch.sync_from_state(request.app.state, _data_dir(request)) # 绑定即评估 return {"rule": rule} @router.post("/auto_rules/{rule_id}/enabled") def set_auto_rule_enabled(request: Request, rule_id: str, enabled: bool, account: str = Query(paper.DEFAULT_ACCOUNT_ID)): - from app.strategy import paper_auto + from app.strategy import paper_auto, paper_watch rule = paper_auto.set_enabled(_data_dir(request), rule_id, enabled, _acc(request, account)) if rule is None: raise HTTPException(status_code=404, detail=f"规则不存在: {rule_id}") + paper_watch.sync_from_state(request.app.state, _data_dir(request)) return {"rule": rule} @router.delete("/auto_rules/{rule_id}") def delete_auto_rule(request: Request, rule_id: str, account: str = Query(paper.DEFAULT_ACCOUNT_ID)): - from app.strategy import paper_auto + from app.strategy import paper_auto, paper_watch if not paper_auto.delete_auto_rule(_data_dir(request), rule_id, _acc(request, account)): raise HTTPException(status_code=404, detail=f"规则不存在: {rule_id}") + paper_watch.sync_from_state(request.app.state, _data_dir(request)) return {"ok": True} @@ -470,4 +476,7 @@ def arena_batch_create(request: Request, body: ArenaBatchModel): created.append({"account": account_id, "name": acc_name, "rule_id": saved["id"]}) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) from e + from app.strategy import paper_watch + + paper_watch.sync_from_state(request.app.state, data_dir) # 绑定即评估 return {"created": created} diff --git a/backend/app/main.py b/backend/app/main.py index 55f59a55..ad0a0d13 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -38,6 +38,7 @@ from app.api import ( pages_admin, paper, pipeline, + position_policies, regime, rps, screener, @@ -447,6 +448,15 @@ async def _application_lifespan(app: FastAPI): logger.info("monitor engine loaded: %d rules", monitor_engine.rule_count) except Exception as e: # noqa: BLE001 logger.warning("monitor engine load failed: %s", e) + # 模拟盘「绑定即评估」: 把账户绑定但无监控规则的策略注入评估循环 + # (set_rules 全量覆盖, 必须在其后注入; 失败不阻断启动)。 + try: + from app.strategy import paper_watch + + synced = paper_watch.sync(store.data_dir, monitor_engine, strategy_engine) + logger.info("paper watch synced at startup: %d strategies", len(synced["watching"])) + except Exception as e: # noqa: BLE001 + logger.warning("paper watch sync failed (ignored): %s", e) app.state.monitor_engine = monitor_engine app.state.sector_monitor_service = sector_monitor_service @@ -648,6 +658,7 @@ app.include_router(abnormal.router) app.include_router(regime.router) app.include_router(analysis.router) app.include_router(pipeline.router) +app.include_router(position_policies.router) app.include_router(data.router) app.include_router(ext_data.router) app.include_router(ext_pages.router) diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 948431f0..6af54db8 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -1218,6 +1218,7 @@ class QuoteService: all_alerts: list[dict] = [] rule_events: list[dict] = [] + paper_watch_events: list[dict] = [] # 模拟盘合成规则的沙盒信号 (只喂模拟盘) engine = None # 通用监控规则评估 (统一引擎: signal/price/market/strategy) @@ -1301,6 +1302,12 @@ class QuoteService: ) except Exception as e: # noqa: BLE001 logger.warning("指数监控评估失败 (不影响股票/ETF 告警): %s", e) + # 模拟盘「绑定即评估」的合成规则事件是沙盒信号: 只喂下面的 + # 模拟盘钩子, 不进告警流 (留痕/SSE/系统通知/webhook/扩展格式化)。 + from app.strategy.paper_watch import is_paper_watch_event + + paper_watch_events = [ev for ev in rule_events if is_paper_watch_event(ev)] + rule_events = [ev for ev in rule_events if not is_paper_watch_event(ev)] if rule_events: rule_events = self._format_extension_notifications(rule_events) # 落盘到 alerts.jsonl @@ -1379,8 +1386,9 @@ class QuoteService: for acc_id in paper_trading.list_account_ids(data_dir): paper_events.extend( paper_trading.evaluate_intraday(data_dir, snapshot, account_id=acc_id)) - if rule_events: - created = paper_auto.on_rule_events(data_dir, rule_events, account_id=acc_id) + feed = rule_events + paper_watch_events + if feed: + created = paper_auto.on_rule_events(data_dir, feed, account_id=acc_id) paper_events.extend(paper_auto.auto_order_events(created, account_id=acc_id)) # 成交/自动跟单下单推送 (V3): 复用监控中心既有管道 —— SSE toast / # 语音 (前端按 source 拼文案) / 系统通知 / alert_store 留痕 / diff --git a/backend/app/strategy/paper_watch.py b/backend/app/strategy/paper_watch.py new file mode 100644 index 00000000..7947fb2d --- /dev/null +++ b/backend/app/strategy/paper_watch.py @@ -0,0 +1,119 @@ +"""模拟盘「绑定即评估」— 把账户绑定的策略自动接入实时评估循环。 + +背景: 模拟盘自动跟单吃的是监控引擎的事件流, 但事件源(策略评估)此前只由 +监控页的用户动作(建 type=strategy 规则)开启 — 「绑定」成了订阅了一个 +不会打开的事件源。本模块在绑定变化时向 MonitorRuleEngine 注入/移除 +**内存态合成规则**(不落盘、不进监控页), 使绑定即生效; 监控页回归本职 +(告警推送), 与模拟盘互不依赖。 + +性能与去重红线: +- 评估引擎仍只有一个, 每策略每轮只评估一次、事件扇出给全部订阅账户; +- 某策略已有启用中的真监控规则时不注入合成规则(双规则会重复推事件); +- 合成事件只喂模拟盘下单, 不进 SSE 告警/系统通知/webhook/留痕 + (quote_service 用 is_paper_watch_event 分流)。 +""" +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +PAPER_WATCH_RULE_PREFIX = "paper_watch:" +# side=auto 跟单需要全部四类事件; 引擎默认只有池事件, 必须显式放开。 +_NOTIFY_EVENTS = ("buy_signal", "sell_signal", "pool_entry", "pool_exit") + + +def is_paper_watch_event(ev: dict) -> bool: + """事件是否来自模拟盘合成规则 (路由分流用: 只喂模拟盘, 不进告警流)。""" + return str(ev.get("rule_id") or "").startswith(PAPER_WATCH_RULE_PREFIX) + + +def synthetic_rule_id(strategy_id: str) -> str: + return f"{PAPER_WATCH_RULE_PREFIX}{strategy_id}" + + +def _make_rule(strategy_id: str) -> dict: + return { + "id": synthetic_rule_id(strategy_id), + "name": f"模拟盘自动评估 · {strategy_id}", + "type": "strategy", + "strategy_id": strategy_id, + "asset_type": "stock", + # scope 默认 "symbols" 且空列表会把候选过滤成空, 必须显式全市场 + "scope": "all", + "enabled": True, + "notify_events": list(_NOTIFY_EVENTS), + } + + +def sync_from_state(app_state: Any, data_dir: Path) -> dict[str, Any] | None: + """API 层便捷入口: 从 app.state 取两个引擎执行 sync。 + + 辅助钩子, 不得影响调用方主流程: 引擎未就绪返回 None; + 引擎形态异常/内部错误只告警不抛出 (单元测试直连 sync() 覆盖真实逻辑)。 + """ + monitor_engine = getattr(app_state, "monitor_engine", None) + strategy_engine = getattr(app_state, "strategy_engine", None) + if monitor_engine is None or strategy_engine is None: + return None + try: + return sync(data_dir, monitor_engine, strategy_engine) + except Exception as e: # noqa: BLE001 + logger.warning("paper watch sync failed (ignored): %s", e) + return None + + +def watched_strategy_ids(data_dir: Path) -> set[str]: + """全部账户「启用的 strategy 型自动跟单规则」绑定的策略 id 集合。""" + from app.strategy import paper, paper_auto + + ids: set[str] = set() + for acc_id in paper.list_account_ids(data_dir): + for rule in paper_auto.load_auto_rules(data_dir, acc_id, enabled_only=True): + if rule.get("match_kind") == "strategy" and rule.get("match_id"): + ids.add(str(rule["match_id"])) + return ids + + +def sync(data_dir: Path, monitor_engine: Any, strategy_engine: Any) -> dict[str, Any]: + """把「未被真监控规则覆盖的绑定策略」增量注入/移出评估循环 (幂等)。 + + 必须在每次 set_rules (全量覆盖会清掉合成规则) 之后重新调用 — + 接线点见 main 启动 / monitor_rules._sync_engine / paper 规则 CRUD / 策略删除。 + """ + covered = { + str(r.get("strategy_id")) + for r in monitor_engine.rules.values() + if r.get("enabled", True) + and r.get("type") == "strategy" + and not str(r.get("id") or "").startswith(PAPER_WATCH_RULE_PREFIX) + } + want: set[str] = set() + for sid in watched_strategy_ids(data_dir): + if sid in covered: + continue + try: + strategy_engine.get(sid) + except ValueError: + continue # 策略已删除/失效, 下次 sync 前引擎侧本就会跳过它 + want.add(sid) + + current = { + str(r.get("strategy_id")) + for rid, r in monitor_engine.rules.items() + if str(rid).startswith(PAPER_WATCH_RULE_PREFIX) + } + added = sorted(want - current) + removed = sorted(current - want) + for sid in added: + monitor_engine.add_rule(_make_rule(sid)) + for sid in removed: + monitor_engine.remove_rule(synthetic_rule_id(sid)) + if added or removed: + logger.info( + "paper watch synced: +%s -%s (watching %d strategies)", + added, removed, len(want), + ) + return {"added": added, "removed": removed, "watching": sorted(want)} diff --git a/backend/tests/test_paper_watch.py b/backend/tests/test_paper_watch.py new file mode 100644 index 00000000..f851f193 --- /dev/null +++ b/backend/tests/test_paper_watch.py @@ -0,0 +1,172 @@ +"""模拟盘「绑定即评估」— paper_watch 同步与合成规则事件流端到端。 + +覆盖: watched 聚合 (多账户/禁用排除/非 strategy 排除); sync 注入/幂等/与真 +监控规则去重/策略已删跳过/解绑移除; 合成规则产生的事件带 paper_watch 前缀 +并能驱动 paper_auto 自动下单 (两轮 evaluate: 首轮基线不刷事件)。 +""" +from __future__ import annotations + +from datetime import date +from types import SimpleNamespace + +import pytest + +from app.strategy import paper, paper_auto, paper_watch +from app.strategy.engine import StrategyResult +from app.strategy.monitor import MonitorRuleEngine + + +class _FakeStrategyEngine: + """只认 demo 策略; run 按序吐结果 (与 tests/test_strategy_monitor_events 同模式)。""" + + def __init__(self, results: list[StrategyResult], known: set[str] | None = None): + self.results = list(results) + self.known = known if known is not None else {"demo"} + self.strategy = SimpleNamespace( + meta={"id": "demo", "name": "示例策略"}, + execution_backend="polars_expr", + filter_history_fn=None, + ) + + def get(self, strategy_id: str): + if strategy_id not in self.known: + raise ValueError(f"unknown strategy: {strategy_id}") + return self.strategy + + def run(self, strategy_id: str, context, **kwargs): + return self.results.pop(0) + + +def _result(pool: tuple[str, ...], buys: tuple[str, ...] = ()) -> StrategyResult: + return StrategyResult( + as_of=date(2026, 10, 27), + strategy_id="demo", + rows=[{"symbol": s, "close": 10.0, "change_pct": 0.01} for s in pool], + total=len(pool), + scores={}, + entry_signal_hits=[{"symbol": s, "signals": ["signal_buy"]} for s in buys], + exit_signal_hits=[], + ) + + +def _quotes(): + import polars as pl + + return pl.DataFrame({ + "symbol": ["A", "B"], + "close": [10.0, 20.0], + "change_pct": [0.01, -0.02], + }) + + +@pytest.fixture() +def paper_env(tmp_path): + """两个账户: acc1 绑 demo (启用), acc2 绑 demo (禁用) + 一条 rule 型规则。""" + paper.create_account(tmp_path, 100_000.0, account_id="acc1", name="一") + paper.create_account(tmp_path, 100_000.0, account_id="acc2", name="二") + paper_auto.create_auto_rule(tmp_path, { + "name": "跟demo", "match_kind": "strategy", "match_id": "demo", + "side": "buy", "size_mode": "fixed_amount", "size_value": 5000, + "order_type": "next_open", "cooldown_days": 5, "max_positions": 0, + "enabled": True, + }, account_id="acc1") + disabled = paper_auto.create_auto_rule(tmp_path, { + "name": "停用", "match_kind": "strategy", "match_id": "demo", + "side": "buy", "size_mode": "fixed_amount", "size_value": 5000, + "order_type": "next_open", "cooldown_days": 5, "max_positions": 0, + "enabled": False, + }, account_id="acc2") + rule_kind = paper_auto.create_auto_rule(tmp_path, { + "name": "跟规则", "match_kind": "rule", "match_id": "mr_x", + "side": "buy", "size_mode": "fixed_amount", "size_value": 5000, + "order_type": "next_open", "cooldown_days": 5, "max_positions": 0, + "enabled": True, + }, account_id="acc2") + return {"disabled_id": disabled["id"], "rule_kind_id": rule_kind["id"]} + + +def test_watched_strategy_ids_aggregates(tmp_path, paper_env): + assert paper_watch.watched_strategy_ids(tmp_path) == {"demo"} + + +def test_sync_injects_synthetic_rule_and_is_idempotent(tmp_path, paper_env): + engine = MonitorRuleEngine() + engine.set_strategy_engine(_FakeStrategyEngine([])) + result = paper_watch.sync(tmp_path, engine, _FakeStrategyEngine([])) + assert result["added"] == ["demo"] + rule = engine.rules["paper_watch:demo"] + assert rule["type"] == "strategy" + assert rule["scope"] == "all" + assert set(rule["notify_events"]) == {"buy_signal", "sell_signal", "pool_entry", "pool_exit"} + assert rule["enabled"] is True + + again = paper_watch.sync(tmp_path, engine, _FakeStrategyEngine([])) + assert again["added"] == [] and again["removed"] == [] + + +def test_sync_dedupes_against_real_monitor_rule(tmp_path, paper_env): + engine = MonitorRuleEngine() + engine.set_strategy_engine(_FakeStrategyEngine([])) + engine.add_rule({ + "id": "mr_real", "name": "真监控", "type": "strategy", "strategy_id": "demo", + "asset_type": "stock", "scope": "all", "enabled": True, + "notify_events": ["pool_entry"], + }) + result = paper_watch.sync(tmp_path, engine, _FakeStrategyEngine([])) + assert result["watching"] == [] + assert "paper_watch:demo" not in engine.rules + + +def test_sync_skips_deleted_strategy(tmp_path, paper_env): + engine = MonitorRuleEngine() + engine.set_strategy_engine(_FakeStrategyEngine([], known=set())) + result = paper_watch.sync(tmp_path, engine, _FakeStrategyEngine([], known=set())) + assert result["watching"] == [] + + +def test_sync_removes_unbound(tmp_path, paper_env): + engine = MonitorRuleEngine() + engine.set_strategy_engine(_FakeStrategyEngine([])) + paper_watch.sync(tmp_path, engine, _FakeStrategyEngine([])) + assert "paper_watch:demo" in engine.rules + paper_auto.delete_auto_rule(tmp_path, _first_rule_id(tmp_path, "acc1"), account_id="acc1") + result = paper_watch.sync(tmp_path, engine, _FakeStrategyEngine([])) + assert result["removed"] == ["demo"] + assert "paper_watch:demo" not in engine.rules + + +def _first_rule_id(data_dir, account_id) -> str: + return paper_auto.load_auto_rules(data_dir, account_id)[0]["id"] + + +def test_synthetic_rule_events_drive_paper_orders(tmp_path, paper_env, monkeypatch): + """端到端: sync 注入 → 两轮 evaluate (首轮基线) → 事件带 paper_watch 前缀 → + paper_auto 按绑定规则建买单; 同时验证公开事件流里看不到沙盒信号。""" + strat = _FakeStrategyEngine([ + _result(pool=("A",)), + _result(pool=("A", "B"), buys=("B",)), + ]) + engine = MonitorRuleEngine() + engine.set_strategy_engine(strat) + paper_watch.sync(tmp_path, engine, strat) + assert engine.rule_count == 1 + + import app.strategy.monitor as monitor_mod + + monkeypatch.setattr(monitor_mod.time, "time", lambda: 100.0) + assert engine.evaluate(_quotes()) == [] # 首轮基线, 不刷事件 + + events = engine.evaluate(_quotes()) + assert events, "第二轮应产生事件" + assert all(paper_watch.is_paper_watch_event(ev) for ev in events) + buys = [ev for ev in events if ev["type"] == "buy_signal"] + assert {ev["symbol"] for ev in buys} == {"B"} + + created = paper_auto.on_rule_events(tmp_path, events, account_id="acc1") + assert len(created) == 1 + order = created[0] + assert order["symbol"] == "B" + assert order["side"] == "buy" + assert order["status"] == "pending" # next_open 次日开盘成交 + # 账户二无启用规则, 不会重复下单 + assert paper_auto.on_rule_events(tmp_path, events, account_id="acc2") == []