feat(paper): V3 补全 — 结算成交留痕/自动跟单下单事件推送/AI 持仓注入多账户

- paper.day_fill_events: 交易日成交台账 → AlertEvent 同构事件;
  daily_pipeline paper_settle 阶段补 alert_store 留痕 (next_open/close
  单的成交发生在盘后管道, 盘中钩子覆盖不到; 管道无 SSE 广播器, 留痕即可见)
- paper_auto.auto_order_events: 自动跟单创建的订单 → 推送事件
  (rule_id 从 order.source 还原), quote_service 钩子随成交事件一并走
  SSE toast / 语音 / 系统通知 / 留痕 / Webhook 既有通道
- stock_analyzer: 虚拟持仓注入改为逐账户收集 (V2 多账户), 空仓账户跳过,
  每账户一行标注账户名; raw 口径浮盈不变
- tests: 新增 day_fill_events / auto_order_events / 多账户注入用例,
  既有注入用例适配列表签名 (66 passed)
This commit is contained in:
shy3130
2026-09-24 20:01:14 +08:00
parent 4db9c7b6fc
commit f249856bf7
7 changed files with 157 additions and 28 deletions
+13
View File
@@ -745,6 +745,19 @@ def run_now(
paper_summary = totals
if paper_summary.get("filled") or paper_summary.get("corp_actions"):
logger.info("paper_settle: %s", paper_summary)
# 结算成交留痕 (V3): next_open/close 单的成交发生在盘后管道内, 盘中钩子
# 覆盖不到; 管道无 SSE 广播器, 这里补 alert_store 留痕进监控中心即可见。
if totals["filled"]:
try:
from app.services import alert_store
settle_events = []
for acc_id in paper_trading.list_account_ids(repo.store.data_dir):
settle_events.extend(paper_trading.day_fill_events(
repo.store.data_dir, today.isoformat(), account_id=acc_id))
if settle_events:
alert_store.append_many(repo.store.data_dir, settle_events)
except Exception as e:
logger.warning("模拟盘结算成交留痕失败: %s", e)
emit("paper_settle", 94, "模拟盘结算完成")
except Exception as e:
logger.warning("paper_settle failed (soft): %s", e)
+5 -3
View File
@@ -1338,9 +1338,11 @@ class QuoteService:
paper_events.extend(
paper_trading.evaluate_intraday(data_dir, snapshot, account_id=acc_id))
if rule_events:
paper_auto.on_rule_events(data_dir, rule_events, account_id=acc_id)
# 成交推送 (V3): 复用监控中心既有管道 —— SSE toast / 语音 (前端按
# source 拼文案) / 系统通知 / alert_store 留痕 / Webhook。全部静默降级。
created = paper_auto.on_rule_events(data_dir, rule_events, account_id=acc_id)
paper_events.extend(paper_auto.auto_order_events(created, account_id=acc_id))
# 成交/自动跟单下单推送 (V3): 复用监控中心既有管道 —— SSE toast /
# 语音 (前端按 source 拼文案) / 系统通知 / alert_store 留痕 /
# Webhook。全部静默降级。
if paper_events:
self._broadcast_alerts(paper_events)
try:
+25 -18
View File
@@ -195,24 +195,27 @@ _SYSTEM_PROMPT = """你是一位拥有 15 年 A 股一线研究经验的技术
# 用户消息构建
# ================================================================
def _paper_position_part(pos: dict | None, close: float | None, raw_close: float | None) -> str:
def _paper_position_part(positions: list[dict], close: float | None, raw_close: float | None) -> str:
"""虚拟持仓上下文 (V3): 模拟盘持有该标的时注入提示词, 让 AI 结合仓位给建议。
pos 为空 / 数量或成本无效时返回空串 (不注入"空仓"之类误导性表述)。
positions 为多账户列表 [{account_id, qty, avg_cost}, ...] (V2 账户隔离),
只收 qty>0 的账户, 逐账户一行; 全空则返回空串 (不注入"空仓"之类误导性表述)。
浮盈用 raw 口径: 持仓 avg_cost 是不复权 raw 成本, 与前复权 close 混算会在
除权后产生虚假盈亏, 故优先用 raw_close (缺列时退化用 close 并接受口径误差)。
"""
if not pos:
lines: list[str] = []
for item in positions or []:
qty = item.get("qty") or 0
avg_cost = item.get("avg_cost") or 0
if qty <= 0 or avg_cost <= 0:
continue
line = f"账户 {item.get('account_id') or 'default'}: 持仓 {qty} 股, 平均成本 {avg_cost:.3f} 元 (不复权)"
ref = raw_close if (raw_close is not None and raw_close > 0) else close
if ref is not None and ref > 0:
line += f", 现价 {ref:.2f} 元, 浮动盈亏 {(ref / avg_cost - 1) * 100:+.2f}%"
lines.append(line)
if not lines:
return ""
qty = pos.get("qty") or 0
avg_cost = pos.get("avg_cost") or 0
if qty <= 0 or avg_cost <= 0:
return ""
ref = raw_close if (raw_close is not None and raw_close > 0) else close
lines = [f"持仓数量: {qty} 股, 平均成本: {avg_cost:.3f} 元 (不复权)"]
if ref is not None and ref > 0:
pnl_pct = (ref / avg_cost - 1) * 100
lines.append(f"现价 {ref:.2f} 元, 浮动盈亏: {pnl_pct:+.2f}%")
return (
"\n以下是该标的在模拟盘 (虚拟账户) 的当前持仓:\n"
+ "\n".join(lines)
@@ -229,7 +232,7 @@ def _build_user_prompt(
symbol: str,
focus: str,
asset_type: str = "stock",
paper_position: dict | None = None,
paper_positions: list[dict] | None = None,
raw_close: float | None = None,
) -> str:
"""构建用户消息:标的 + 价位摘要 + 技术指标 JSON + 财务摘要 + 关注点 (+ 虚拟持仓)。
@@ -282,7 +285,7 @@ def _build_user_prompt(
focus_instruction = build_focus_instruction(focus, report_name="个股分析报告")
if focus_instruction:
parts.extend(["", focus_instruction])
paper_part = _paper_position_part(paper_position, close, raw_close)
paper_part = _paper_position_part(paper_positions or [], close, raw_close)
if paper_part:
parts.extend(["", paper_part])
return "\n".join(parts)
@@ -356,11 +359,15 @@ async def analyze_stock_stream(
try:
from app.services.ai_provider import stream_ai_text
# 虚拟持仓上下文 (V3): 默认模拟盘账户持有该标的时注入 (任何异常静默跳过, 不影响分析)。
paper_position = None
# 虚拟持仓上下文 (V3): 逐账户 (V2 多账户) 收集持有该标的的持仓注入
# (任何异常静默跳过, 不影响分析)。
paper_positions: list[dict] = []
try:
from app.strategy import paper as paper_trading
paper_position = paper_trading.load_positions(data_dir).get(symbol)
for acc_id in paper_trading.list_account_ids(data_dir):
pos = paper_trading.load_positions(data_dir, account_id=acc_id).get(symbol)
if pos and (pos.get("qty") or 0) > 0:
paper_positions.append({"account_id": acc_id, **pos})
except Exception as e: # noqa: BLE001
logger.warning("读取模拟盘持仓失败 (跳过注入): %s", e)
# 浮盈口径用 raw: 持仓成本不复权, 与前复权 close 混算会在除权后失真
@@ -368,7 +375,7 @@ async def analyze_stock_stream(
kline_tail = _clean_rows(df, _KLINE_KEEP_COLS)
user_prompt = _build_user_prompt(kline_tail, fins, levels, close, symbol, focus,
asset_type=asset_type, paper_position=paper_position,
asset_type=asset_type, paper_positions=paper_positions,
raw_close=raw_close)
got_content = False
async for delta in stream_ai_text(
+13
View File
@@ -713,6 +713,19 @@ def _queue_or_expire(data_dir: Path, order: dict, acc: dict, reason: str, accoun
_expire(data_dir, order, reason, account_id)
def day_fill_events(data_dir: Path, day: str, account_id: str = DEFAULT_ACCOUNT_ID) -> list[dict]:
"""指定交易日的全部成交 → 推送事件。
盘后结算路径 (daily_pipeline paper_settle) 用它把结算成交留痕到告警中心;
盘中即时成交走 evaluate_intraday 返回值, 不经过本函数。
"""
return [
_fill_event(f, account_id)
for f in load_fills(data_dir, account_id)
if f.get("date") == day and f.get("kind") == "fill"
]
def _fill_event(fill: dict, account_id: str) -> dict:
"""成交台账行 → 推送事件。字段对齐监控告警 (AlertEvent), 供 SSE toast /
语音播报 / 系统通知 / alert_store 留痕 / Webhook 复用, 前端无需新管道。
+28
View File
@@ -201,3 +201,31 @@ def on_rule_events(data_dir: Path, events: list[dict], account_id: str = paper.D
created.append(order)
logger.info("paper auto %s: %s 触发 %s %d 股 (%s)", rule["name"], symbol, rule["side"], qty, order["id"])
return created
_ORDER_TYPE_LABEL = {"market": "即时", "next_open": "次日开盘", "close": "当日收盘"}
def auto_order_events(created: list[dict], account_id: str = paper.DEFAULT_ACCOUNT_ID) -> list[dict]:
"""自动跟单创建的订单 → 推送事件。与成交事件同构 (source=paper 对齐
AlertEvent), 走同一推送通道; rule_id 从 order.source (auto:{rule_id}) 还原。
"""
events: list[dict] = []
for o in created:
source = str(o.get("source", ""))
rule_id = source.removeprefix("auto:") if source.startswith("auto:") else ""
side_label = "买入" if o["side"] == "buy" else "卖出"
type_label = _ORDER_TYPE_LABEL.get(o["order_type"], o["order_type"])
events.append({
"source": "paper",
"type": "auto_order",
"severity": "info",
"ts": int(cn_now().timestamp() * 1000),
"symbol": o["symbol"],
"rule_id": rule_id,
"side": o["side"],
"qty": o["qty"],
"account_id": account_id,
"message": f"自动跟单触发{side_label}下单 ({type_label}) {o['qty']}股",
})
return events
+48
View File
@@ -628,3 +628,51 @@ def test_legacy_single_account_layout_migrates(tmp_path, monkeypatch):
assert not (legacy / "account.json").exists()
# 幂等: 重复访问不报错
assert paper.get_account(tmp_path)["cash"] == 750000.0
# ── V3 补全: 结算成交留痕 / 自动跟单下单事件 ──────────
def test_day_fill_events_returns_todays_fills(tmp_path, monkeypatch):
"""day_fill_events: 只取指定交易日、kind=fill 的台账行, 转 AlertEvent 同构事件。"""
day = date(2026, 9, 24)
monkeypatch.setattr(paper, "cn_today", lambda: day)
_write_daily(tmp_path, [(day - timedelta(days=1), 10.0, 10.0)])
_cap_account(tmp_path)
# 当日无成交 → 空列表
assert paper.day_fill_events(tmp_path, day.isoformat()) == []
_, err = paper.create_order(tmp_path, SYM, "buy", qty=1000, ref_price=10.0)
assert err is None
assert len(paper.evaluate_intraday(tmp_path, {SYM: 10.0})) == 1
events = paper.day_fill_events(tmp_path, day.isoformat())
assert len(events) == 1
ev = events[0]
assert ev["source"] == "paper" and ev["type"] == "fill" and ev["severity"] == "info"
assert ev["symbol"] == SYM and ev["side"] == "buy" and ev["qty"] == 1000
assert isinstance(ev["ts"], int) and ev["ts"] > 0
assert "买入成交" in ev["message"]
# 幂等 (结算重跑同日不重复追加事件)
assert paper.day_fill_events(tmp_path, day.isoformat()) == events
def test_auto_order_events_shape(tmp_path):
"""auto_order_events: 订单 → AlertEvent 同构事件, rule_id 从 source 还原。"""
from app.strategy import paper_auto
orders = [
{"symbol": "600000.SH", "side": "buy", "qty": 500,
"order_type": "next_open", "source": "auto:rule_abc"},
{"symbol": "000001.SZ", "side": "sell", "qty": 200,
"order_type": "market", "source": "manual"},
]
events = paper_auto.auto_order_events(orders, account_id="acc1")
assert len(events) == 2
buy_ev, sell_ev = events
assert buy_ev["source"] == "paper" and buy_ev["type"] == "auto_order"
assert buy_ev["rule_id"] == "rule_abc" and buy_ev["account_id"] == "acc1"
assert "买入" in buy_ev["message"] and "次日开盘" in buy_ev["message"] and "500股" in buy_ev["message"]
assert isinstance(buy_ev["ts"], int) and buy_ev["ts"] > 0
assert sell_ev["rule_id"] == "" and "卖出" in sell_ev["message"] and "即时" in sell_ev["message"]
# 空列表 → 空
assert paper_auto.auto_order_events([]) == []
+25 -7
View File
@@ -37,7 +37,8 @@ def test_user_prompt_injects_paper_position():
"""模拟盘持有该标的: 注入持仓数量/成本/浮盈, 并要求 AI 给持仓建议。"""
from app.services.stock_analyzer import _paper_position_part, _build_user_prompt
part = _paper_position_part({"qty": 1000, "avg_cost": 10.0}, close=11.0, raw_close=11.0)
part = _paper_position_part([{"account_id": "default", "qty": 1000, "avg_cost": 10.0}],
close=11.0, raw_close=11.0)
assert "1000 股" in part
assert "10.000" in part
assert "+10.00%" in part
@@ -47,18 +48,35 @@ def test_user_prompt_injects_paper_position():
kline_tail=[{"date": "2026-07-24", "close": 11.0}],
fins={"metrics": [], "income": []},
levels={}, close=11.0, symbol="600000.SH", focus="",
paper_position={"qty": 1000, "avg_cost": 10.0}, raw_close=11.0,
paper_positions=[{"account_id": "default", "qty": 1000, "avg_cost": 10.0}],
raw_close=11.0,
)
assert "模拟盘" in prompt and "持仓数量" in prompt
assert "模拟盘" in prompt and "持仓 1000 股" in prompt
def test_paper_position_part_multi_account():
"""V2 多账户: 逐账户注入, 空仓账户跳过, 全空返回空串。"""
from app.services.stock_analyzer import _paper_position_part
part = _paper_position_part(
[
{"account_id": "a1", "qty": 100, "avg_cost": 10.0},
{"account_id": "a2", "qty": 0, "avg_cost": 10.0},
{"account_id": "a3", "qty": 200, "avg_cost": 20.0},
],
close=11.0, raw_close=11.0,
)
assert "账户 a1" in part and "账户 a3" in part
assert "a2" not in part
def test_paper_position_part_empty_without_position():
"""无持仓/零仓/无效成本: 返回空串, 不注入误导性「空仓」表述。"""
from app.services.stock_analyzer import _paper_position_part
assert _paper_position_part(None, 11.0, 11.0) == ""
assert _paper_position_part({"qty": 0, "avg_cost": 10.0}, 11.0, 11.0) == ""
assert _paper_position_part({"qty": 100, "avg_cost": 0}, 11.0, 11.0) == ""
assert _paper_position_part([], 11.0, 11.0) == ""
assert _paper_position_part([{"qty": 0, "avg_cost": 10.0}], 11.0, 11.0) == ""
assert _paper_position_part([{"qty": 100, "avg_cost": 0}], 11.0, 11.0) == ""
def test_paper_position_part_prefers_raw_close():
@@ -68,6 +86,6 @@ def test_paper_position_part_prefers_raw_close():
"""
from app.services.stock_analyzer import _paper_position_part
part = _paper_position_part({"qty": 100, "avg_cost": 20.0}, close=11.0, raw_close=21.0)
part = _paper_position_part([{"qty": 100, "avg_cost": 20.0}], close=11.0, raw_close=21.0)
assert "+5.00%" in part
assert "-45" not in part