fix: K线worker限速+新浪日线备选+失败退避（应对东财限速）

deploy/a-stock-timeline.service

@@ -0,0 +1,19 @@
+[Unit]
+Description=JQuant A股时间线实时采集与分析服务（B/S）
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+# 部署路径按实际调整（git clone linux-web 分支后的目录）
+WorkingDirectory=/opt/a_stock_timeline
+ExecStart=/opt/a_stock_timeline/venv/bin/python -m src.web.server
+Restart=always
+RestartSec=5
+# 内部单机部署以 root 运行；生产建议改为专用低权限用户并收紧数据目录权限
+User=root
+Environment=PORT=8100
+# 日志进 journald：journalctl -u a-stock-timeline -f
+
+[Install]
+WantedBy=multi-user.target

deploy/e2e_server_test.py

@@ -0,0 +1,29 @@
+# -*- coding: utf-8 -*-
+"""服务器端端到端验证：宇宙 40 只 → K线同步 → 打分 → 推荐卡（在服务器 venv 内执行）"""
+import sys, warnings, time
+sys.path.insert(0, '/opt/a_stock_timeline')
+warnings.filterwarnings('ignore')
+from src.fetcher.kline_fetcher import KlineFetcher
+from src.quant.engine import QuantEngine
+
+DB = '/opt/a_stock_timeline/data/a_stock.db'
+eng = QuantEngine(emit=lambda e: print('[emit]', e['kind'], flush=True), db_path=DB)
+codes, names = eng.universe(40)
+print('宇宙股票:', len(codes), flush=True)
+kf = KlineFetcher(DB)
+t0 = time.time()
+ok = 0
+for i, c in enumerate(codes):
+    ok += 1 if kf.sync(c, 'day', days=150) else 0
+    if (i + 1) % 10 == 0:
+        print('  进度 {}/{} 成功{} 用时{:.0f}s'.format(i + 1, len(codes), ok, time.time() - t0), flush=True)
+print('K线同步完成: 成功 {} 只，用时 {:.0f}s'.format(ok, time.time() - t0), flush=True)
+recs = eng.build_recommendations(10)
+print('=== 推荐卡（Top 10）===', flush=True)
+for r in recs:
+    print('★{stars} {name}({code}) 现价{price} 购入{buy_low}~{buy_high} 预计{er}% | {reason}'.format(
+        stars=r['stars'], name=r['name'], code=r['code'], price=r['price'],
+        buy_low=r['buy_low'], buy_high=r['buy_high'],
+        er=r['expected_return_pct'] if r['expected_return_pct'] is not None else '—',
+        reason=r['reason']), flush=True)
+print('E2E DONE', flush=True)

deploy/probe_server.py

@@ -0,0 +1,28 @@
+# -*- coding: utf-8 -*-
+"""探测目标服务器环境（SSH 只读探测，不做变更）"""
+import paramiko
+
+HOST, USER, PWD = '10.20.226.110', 'root', 'tzt123@123'
+
+cli = paramiko.SSHClient()
+cli.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+cli.connect(HOST, username=USER, password=PWD, timeout=10)
+
+CMDS = [
+    ('系统', 'cat /etc/os-release 2>/dev/null | head -2 || uname -a'),
+    ('内核', 'uname -r'),
+    ('python3', 'python3 --version 2>&1; which python3'),
+    ('git', 'git --version 2>&1'),
+    ('venv模块', 'python3 -m venv --help >/dev/null 2>&1 && echo venv-ok || echo venv-missing'),
+    ('防火墙', 'systemctl is-active firewalld 2>/dev/null; systemctl is-active ufw 2>/dev/null; iptables -L INPUT -n 2>/dev/null | head -3'),
+    ('8100占用', 'ss -tlnp 2>/dev/null | grep 8100 || echo 空闲'),
+    ('/opt可写', 'test -d /opt && echo opt-exists; touch /opt/.wtest 2>/dev/null && rm /opt/.wtest && echo opt-writable'),
+    ('外网(pypi)', 'curl -s -o /dev/null -w "%{http_code}" --max-time 8 https://pypi.tuna.tsinghua.edu.cn/simple/ || echo unreachable'),
+]
+for label, cmd in CMDS:
+    _, out, err = cli.exec_command(cmd, timeout=20)
+    o = out.read().decode('utf-8', 'ignore').strip()
+    e = err.read().decode('utf-8', 'ignore').strip()
+    print('【{}】{}'.format(label, o or e or '(空)'))
+
+cli.close()

docs/DEPLOY_LINUX.md

@@ -0,0 +1,73 @@
+# Linux B/S 部署（linux-web 分支）
+
+架构：Linux 服务器常驻执行「实时抓取 + 分析」（采集引擎每 60s 一轮），
+分析事件经 **WebSocket** 实时推送到所有已连接的 Web 仪表盘（浏览器打开即看，无需刷新）。
+
+```
+采集引擎 TimelineCollector（后台线程）
+  ├─ 盘前：美股隔夜收盘 → 实证先验（skill）预估今日跳空/日内概率
+  ├─ 盘中：东财7x24快讯入库 → 关键词告警 → 上证实时点位
+  └─ 盘后：龙虎榜入库 → 收盘日报
+        │ 事件回调（每条：新快讯/盘中点位/告警/日报…）
+        ▼
+aiohttp 服务（端口 8100）
+  ├─ GET  /            Web 仪表盘（深色，事件流实时上屏）
+  ├─ GET  /api/recent  最近事件 JSON
+  ├─ GET  /api/stats   连接数/缓冲统计
+  └─ WS   /ws          实时广播（断线自动重连）
+```
+
+## 部署步骤
+
+```bash
+sudo mkdir -p /opt && cd /opt
+sudo git clone -b linux-web <仓库地址> a_stock_timeline
+cd a_stock_timeline
+
+python3 -m venv venv
+venv/bin/pip install -r requirements.txt
+
+sudo cp deploy/a-stock-timeline.service /etc/systemd/system/
+sudo systemctl daemon-reload
+sudo systemctl enable --now a-stock-timeline
+
+# 查看实时日志
+journalctl -u a-stock-timeline -f
+```
+
+浏览器访问 `http://<服务器IP>:8100`（生产建议前置 nginx 做 TLS/域名）。
+
+## 数据说明
+
+- 数据库：`/opt/a_stock_timeline/data/a_stock.db`（SQLite，与 C 端同 schema）
+- 事件流水：`data/realtime_feed.jsonl`（重启后自动回放最近 200 条到新连接的客户端）
+- 实证先验：`docs/overnight_transmission.md`（由 windows-desktop 分支的
+  `src/analysis/mine_patterns.py` 挖掘生成，拷贝到 docs/ 即可被服务端加载）
+
+## 与 C 端（windows-desktop 分支）的关系
+
+- 共享：fetcher/storage/分析引擎与全部实证口径
+- 差异：C 端入口为 `start_realtime.bat` + 控制台输出 + Windows 数据路径；
+  linux-web 端入口为 `python -m src.web.server`，事件走 WebSocket 广播，路径全部相对化
+
+## 启用「事件影响分析 → 个股推荐」（需大模型）
+
+服务端设置任意一家 OpenAI 兼容厂商的 Key（默认智谱 GLM）：
+
+```bash
+# 编辑服务单元，在 [Service] 段加环境变量
+sudo systemctl edit a-stock-timeline
+# 写入：
+#   [Service]
+#   Environment=JQUANT_LLM_API_KEY=你的Key
+#   Environment=JQUANT_LLM_MODEL=glm-4-flash        # 可选
+#   Environment=JQUANT_LLM_BASE_URL=https://open.bigmodel.cn/api/paas/v4   # 可选
+sudo systemctl restart a-stock-timeline
+```
+
+启用后：盘中出现新快讯 → 自动做「事件影响力分析」→ 页面推送**事件影响卡**
+（事件摘要/方向/板块/置信度 + 推荐个股表：名称、代码、购入区间、预计收益(推测)、
+推荐指数★、简易原因）。未配 Key 时该功能静默待机，其余功能不受影响。
+
+> 推荐卡中的"预计收益"是模型基于历史先验与资金面的**推测值**，页面已标注
+> "模型推断仅供参考，不构成投资建议"。

docs/overnight_transmission.md

@@ -0,0 +1,51 @@
+## 隔夜传导矩阵（美股 -> A股）
+
+> 口径：美股T日涨跌幅分箱 -> A股T+1交易日（跳空=开盘/前收-1；日内=收盘/开盘-1）。样本 2005-2026（受 A 股指数历史与纳斯达克 2014 起数据限制）。
+
+### 纳斯达克100T日 -> 上证指数T+1日
+
+- 美股跌>1.5%（305天）: A股跳空 N=305, 均值 -0.64%, 中位数 -0.41%, 胜率 10.5%; 日内 N=305, 均值 0.20%, 中位数 0.15%, 胜率 56.4%; 全天 N=305, 均值 -0.44%, 中位数 -0.25%, 胜率 37.0%
+- 美股跌0.5~1.5%（484天）: A股跳空 N=484, 均值 -0.20%, 中位数 -0.18%, 胜率 24.6%; 日内 N=484, 均值 0.06%, 中位数 0.10%, 胜率 56.4%; 全天 N=484, 均值 -0.14%, 中位数 -0.06%, 胜率 44.6%
+- 美股正负0.5%内（1287天）: A股跳空 N=1287, 均值 -0.07%, 中位数 -0.06%, 胜率 36.3%; 日内 N=1287, 均值 0.15%, 中位数 0.14%, 胜率 58.7%; 全天 N=1287, 均值 0.08%, 中位数 0.07%, 胜率 53.9%
+- 美股涨0.5~1.5%（737天）: A股跳空 N=737, 均值 0.07%, 中位数 0.04%, 胜率 56.3%; 日内 N=737, 均值 0.08%, 中位数 0.06%, 胜率 52.8%; 全天 N=737, 均值 0.15%, 中位数 0.09%, 胜率 56.3%
+- 美股涨>1.5%（309天）: A股跳空 N=309, 均值 0.22%, 中位数 0.17%, 胜率 73.8%; 日内 N=309, 均值 0.02%, 中位数 0.05%, 胜率 53.4%; 全天 N=309, 均值 0.24%, 中位数 0.23%, 胜率 61.5%
+
+### 纳斯达克100T日 -> 沪深300T+1日
+
+- 美股跌>1.5%（305天）: A股跳空 N=305, 均值 -0.69%, 中位数 -0.50%, 胜率 13.8%; 日内 N=305, 均值 0.23%, 中位数 0.13%, 胜率 56.4%; 全天 N=305, 均值 -0.47%, 中位数 -0.34%, 胜率 35.1%
+- 美股跌0.5~1.5%（484天）: A股跳空 N=484, 均值 -0.20%, 中位数 -0.20%, 胜率 24.2%; 日内 N=484, 均值 0.04%, 中位数 0.04%, 胜率 51.7%; 全天 N=484, 均值 -0.16%, 中位数 -0.14%, 胜率 41.5%
+- 美股正负0.5%内（1287天）: A股跳空 N=1287, 均值 -0.04%, 中位数 -0.05%, 胜率 42.0%; 日内 N=1287, 均值 0.13%, 中位数 0.10%, 胜率 55.7%; 全天 N=1287, 均值 0.09%, 中位数 0.05%, 胜率 52.5%
+- 美股涨0.5~1.5%（737天）: A股跳空 N=737, 均值 0.13%, 中位数 0.07%, 胜率 63.0%; 日内 N=737, 均值 0.04%, 中位数 0.01%, 胜率 50.5%; 全天 N=737, 均值 0.17%, 中位数 0.10%, 胜率 55.0%
+- 美股涨>1.5%（309天）: A股跳空 N=309, 均值 0.32%, 中位数 0.25%, 胜率 79.0%; 日内 N=309, 均值 -0.06%, 中位数 -0.08%, 胜率 45.6%; 全天 N=309, 均值 0.26%, 中位数 0.21%, 胜率 61.2%
+
+### 标普500T日 -> 上证指数T+1日
+
+- 美股跌>1.5%（366天）: A股跳空 N=366, 均值 -0.92%, 中位数 -0.72%, 胜率 7.7%; 日内 N=366, 均值 0.27%, 中位数 0.28%, 胜率 57.9%; 全天 N=366, 均值 -0.65%, 中位数 -0.40%, 胜率 36.1%
+- 美股跌0.5~1.5%（868天）: A股跳空 N=868, 均值 -0.23%, 中位数 -0.21%, 胜率 23.6%; 日内 N=868, 均值 0.02%, 中位数 0.08%, 胜率 53.8%; 全天 N=868, 均值 -0.22%, 中位数 -0.16%, 胜率 43.2%
+- 美股正负0.5%内（2804天）: A股跳空 N=2804, 均值 -0.06%, 中位数 -0.05%, 胜率 39.2%; 日内 N=2804, 均值 0.13%, 中位数 0.13%, 胜率 56.0%; 全天 N=2804, 均值 0.07%, 中位数 0.06%, 胜率 53.2%
+- 美股涨0.5~1.5%（1218天）: A股跳空 N=1218, 均值 0.08%, 中位数 0.06%, 胜率 61.7%; 日内 N=1218, 均值 0.10%, 中位数 0.10%, 胜率 55.1%; 全天 N=1218, 均值 0.19%, 中位数 0.15%, 胜率 59.2%
+- 美股涨>1.5%（319天）: A股跳空 N=319, 均值 0.53%, 中位数 0.37%, 胜率 83.4%; 日内 N=319, 均值 -0.09%, 中位数 -0.09%, 胜率 46.1%; 全天 N=319, 均值 0.44%, 中位数 0.28%, 胜率 62.1%
+
+### 标普500T日 -> 沪深300T+1日
+
+- 美股跌>1.5%（366天）: A股跳空 N=366, 均值 -0.98%, 中位数 -0.77%, 胜率 9.0%; 日内 N=366, 均值 0.35%, 中位数 0.21%, 胜率 57.1%; 全天 N=366, 均值 -0.64%, 中位数 -0.49%, 胜率 38.0%
+- 美股跌0.5~1.5%（868天）: A股跳空 N=868, 均值 -0.25%, 中位数 -0.25%, 胜率 24.9%; 日内 N=868, 均值 0.03%, 中位数 0.00%, 胜率 49.0%; 全天 N=868, 均值 -0.22%, 中位数 -0.20%, 胜率 42.1%
+- 美股正负0.5%内（2804天）: A股跳空 N=2804, 均值 -0.04%, 中位数 -0.04%, 胜率 43.8%; 日内 N=2804, 均值 0.13%, 中位数 0.04%, 胜率 51.9%; 全天 N=2804, 均值 0.08%, 中位数 0.06%, 胜率 52.6%
+- 美股涨0.5~1.5%（1218天）: A股跳空 N=1218, 均值 0.13%, 中位数 0.11%, 胜率 65.4%; 日内 N=1218, 均值 0.08%, 中位数 0.02%, 胜率 50.5%; 全天 N=1218, 均值 0.21%, 中位数 0.15%, 胜率 57.1%
+- 美股涨>1.5%（319天）: A股跳空 N=319, 均值 0.61%, 中位数 0.44%, 胜率 85.9%; 日内 N=319, 均值 -0.18%, 中位数 -0.17%, 胜率 40.8%; 全天 N=319, 均值 0.43%, 中位数 0.32%, 胜率 62.7%
+
+### 道琼斯T日 -> 上证指数T+1日
+
+- 美股跌>1.5%（324天）: A股跳空 N=324, 均值 -0.95%, 中位数 -0.78%, 胜率 8.6%; 日内 N=324, 均值 0.32%, 中位数 0.29%, 胜率 59.0%; 全天 N=324, 均值 -0.64%, 中位数 -0.43%, 胜率 37.0%
+- 美股跌0.5~1.5%（890天）: A股跳空 N=890, 均值 -0.24%, 中位数 -0.20%, 胜率 22.6%; 日内 N=890, 均值 0.11%, 中位数 0.13%, 胜率 56.4%; 全天 N=890, 均值 -0.14%, 中位数 -0.06%, 胜率 45.1%
+- 美股正负0.5%内（2894天）: A股跳空 N=2894, 均值 -0.06%, 中位数 -0.05%, 胜率 39.9%; 日内 N=2894, 均值 0.09%, 中位数 0.10%, 胜率 54.8%; 全天 N=2894, 均值 0.03%, 中位数 0.05%, 胜率 52.5%
+- 美股涨0.5~1.5%（1180天）: A股跳空 N=1180, 均值 0.08%, 中位数 0.07%, 胜率 61.4%; 日内 N=1180, 均值 0.14%, 中位数 0.12%, 胜率 55.7%; 全天 N=1180, 均值 0.23%, 中位数 0.17%, 胜率 59.8%
+- 美股涨>1.5%（287天）: A股跳空 N=287, 均值 0.57%, 中位数 0.37%, 胜率 84.0%; 日内 N=287, 均值 -0.11%, 中位数 -0.14%, 胜率 44.9%; 全天 N=287, 均值 0.45%, 中位数 0.28%, 胜率 60.3%
+
+### 道琼斯T日 -> 沪深300T+1日
+
+- 美股跌>1.5%（324天）: A股跳空 N=324, 均值 -1.01%, 中位数 -0.83%, 胜率 9.6%; 日内 N=324, 均值 0.39%, 中位数 0.20%, 胜率 57.7%; 全天 N=324, 均值 -0.63%, 中位数 -0.49%, 胜率 37.7%
+- 美股跌0.5~1.5%（890天）: A股跳空 N=890, 均值 -0.26%, 中位数 -0.23%, 胜率 24.3%; 日内 N=890, 均值 0.13%, 中位数 0.04%, 胜率 51.9%; 全天 N=890, 均值 -0.13%, 中位数 -0.10%, 胜率 45.2%
+- 美股正负0.5%内（2894天）: A股跳空 N=2894, 均值 -0.04%, 中位数 -0.04%, 胜率 44.7%; 日内 N=2894, 均值 0.07%, 中位数 0.01%, 胜率 50.4%; 全天 N=2894, 均值 0.03%, 中位数 0.03%, 胜率 51.5%
+- 美股涨0.5~1.5%（1180天）: A股跳空 N=1180, 均值 0.12%, 中位数 0.11%, 胜率 64.8%; 日内 N=1180, 均值 0.13%, 中位数 0.04%, 胜率 51.2%; 全天 N=1180, 均值 0.25%, 中位数 0.17%, 胜率 58.1%
+- 美股涨>1.5%（287天）: A股跳空 N=287, 均值 0.64%, 中位数 0.45%, 胜率 83.6%; 日内 N=287, 均值 -0.19%, 中位数 -0.19%, 胜率 42.2%; 全天 N=287, 均值 0.45%, 中位数 0.32%, 胜率 61.0%

patch_frontend_quant.py

@@ -0,0 +1,89 @@
+# -*- coding: utf-8 -*-
+"""前端追加「量化推荐」区（筛选按钮 + 事件渲染 + 表格 + 权重展示）"""
+import io
+
+p = r'src\web\index.html'
+s = io.open(p, encoding='utf-8').read()
+
+# 1) 筛选按钮
+old1 = '    <button data-f="事件影响分析">事件影响</button>'
+new1 = (old1 + '\n    <button data-f="量化推荐">量化推荐</button>\n'
+        '    <button data-f="模型微调">微调</button>')
+assert old1 in s, 'filters not found'
+s = s.replace(old1, new1)
+
+# 2) 主区追加量化推荐表格
+old2 = '  <ul id="feed"><li class="empty">等待服务端推送…</li></ul>\n</main>'
+new2 = ('  <ul id="feed"><li class="empty">等待服务端推送…</li></ul>\n</main>\n'
+        '<section style="max-width:1100px;margin:0 auto 30px">\n'
+        '  <h2 style="font-size:15px;color:var(--text)">量化模型 · 自适应推荐'
+        '（多因子 + IC 微调）</h2>\n'
+        '  <div style="font-size:12px;color:var(--text2);margin-bottom:8px">'
+        '权重：<span id="qw">加载中…</span></div>\n'
+        '  <div style="background:var(--surface);border:1px solid var(--border);'
+        'border-radius:8px;overflow:auto">\n'
+        '    <table style="width:100%;border-collapse:collapse;font-size:13px" id="qtable">\n'
+        '      <thead><tr style="color:var(--text2);text-align:left">\n'
+        '        <th style="padding:8px 10px">代码</th><th style="padding:8px 10px">名称</th>\n'
+        '        <th style="padding:8px 10px">现价</th><th style="padding:8px 10px">模型分</th>\n'
+        '        <th style="padding:8px 10px">购入区间</th>'
+        '<th style="padding:8px 10px">预计收益(回测口径)</th>\n'
+        '        <th style="padding:8px 10px">推荐指数</th>'
+        '<th style="padding:8px 10px">简易原因</th>\n'
+        '      </tr></thead><tbody></tbody>\n'
+        '    </table>\n'
+        '  </div>\n'
+        '  <div style="margin-top:6px;font-size:11px;color:#6b7280">'
+        '⚠ 购入区间与预计收益为模型回测/推测口径，不构成投资建议。</div>\n'
+        '</section>')
+assert old2 in s, 'feed ul not found'
+s = s.replace(old2, new2)
+
+# 3) WS 事件渲染分支（量化推荐）
+old3 = ("  } else {\n"
+        "    const txt = typeof d === 'object' ? JSON.stringify(d) : String(d)\n"
+        "    li.innerHTML = `<span class=\"t\">${e.ts}</span><span class=\"k ${e.kind}\">${e.kind}"
+        "</span><span class=\"d\">${esc(txt)}</span>`\n  }")
+new3 = ("  } else if (e.kind === '量化推荐') {\n"
+        "    renderQuant(e.data)\n"
+        "    const txt = '自适应模型推送 ' + ((d.list || []).length) + ' 只推荐'\n"
+        "    li.innerHTML = `<span class=\"t\">${e.ts}</span><span class=\"k ${e.kind}\">${e.kind}"
+        "</span><span class=\"d\">${esc(txt)}</span>`\n"
+        "  } else {\n"
+        "    const txt = typeof d === 'object' ? JSON.stringify(d) : String(d)\n"
+        "    li.innerHTML = `<span class=\"t\">${e.ts}</span><span class=\"k ${e.kind}\">${e.kind}"
+        "</span><span class=\"d\">${esc(txt)}</span>`\n  }")
+assert old3 in s, 'render branch not found'
+s = s.replace(old3, new3)
+
+# 4) 渲染函数与加载
+old4 = 'connect()'
+new4 = (
+    "function renderQuant(d) {\n"
+    "  const tb = document.querySelector('#qtable tbody')\n"
+    "  if (!tb) return\n"
+    "  tb.innerHTML = (d.list || []).map(r =>\n"
+    "    '<tr><td style=\"padding:6px 10px\">' + esc(r.code) + '</td>' +\n"
+    "    '<td style=\"padding:6px 10px\">' + esc(r.name) + '</td>' +\n"
+    "    '<td style=\"padding:6px 10px\">' + r.price + '</td>' +\n"
+    "    '<td style=\"padding:6px 10px\">' + r.score + '</td>' +\n"
+    "    '<td style=\"padding:6px 10px\">' + r.buy_low + ' ~ ' + r.buy_high + '</td>' +\n"
+    "    '<td style=\"padding:6px 10px\">' + (r.expected_return_pct == null ? '—' : r.expected_return_pct + '%') + '</td>' +\n"
+    "    '<td style=\"padding:6px 10px\" class=\"stars\">' + '★'.repeat(r.stars) + '</td>' +\n"
+    "    '<td style=\"padding:6px 10px\">' + esc(r.reason) + '</td></tr>').join('')\n"
+    "}\n"
+    "function loadQuant() {\n"
+    "  fetch('/api/quant/recommendations').then(r => r.json()).then(renderQuant).catch(() => {})\n"
+    "  fetch('/api/quant/weights').then(r => r.json()).then(w => {\n"
+    "    document.getElementById('qw').textContent =\n"
+    "      Object.entries(w || {}).map(([k, v]) => k + '=' + Number(v).toFixed(2)).join('  ') || '—'\n"
+    "  }).catch(() => {})\n"
+    "}\n"
+    "loadQuant()\n"
+    "setInterval(loadQuant, 60000)\n"
+    "connect()")
+assert old4 in s
+s = s.replace(old4, new4, 1)
+
+io.open(p, 'w', encoding='utf-8').write(s)
+print('index.html 量化推荐区 OK')

requirements.txt

@@ -0,0 +1,8 @@
+akshare>=1.18.92
+pandas>=2.0
+aiohttp>=3.9
+lxml
+beautifulsoup4
+py_mini_racer
+tqdm
+requests

src/analysis/event_impact.py

@@ -0,0 +1,271 @@
+# -*- coding: utf-8 -*-
+"""
+事件影响力分析：新闻/快讯 → 市场影响推断 → 个股推荐卡。
+LLM 只负责解读文本与推断方向；候选股必须落在真实行情快照内（grounding），
+购入区间基于真实现价计算，预计收益标注为推测。
+"""
+import json
+import sqlite3
+import traceback
+from datetime import datetime, timedelta
+
+import requests
+
+from src.analysis.llm_client import LlmClient, LlmError
+
+SYSTEM_PROMPT = """你是 A 股事件影响分析师。系统提供：若干条最新财经快讯/新闻、当日主力资金流入TOP10、\
+今日龙虎榜摘要、上证指数状态，以及全市场个股现价快照（部分）。
+任务：判断这些消息中是否包含值得关注的**事件**；若有，推断该事件对哪个品类（板块）的股票\
+产生什么方向的影响，并从"现价快照"给出的股票中选出最可能受益的标的。
+
+铁律：
+1. 只允许从系统提供的现价快照中选股票，禁止编造代码/名称。
+2. 证据必须引用所给新闻原文片段。
+3. 没有值得分析的事件时输出 has_event=false。
+4. 输出严格 JSON，无 markdown 代码块：
+{"has_event": true|false,
+ "event_summary": "≤80字事件摘要",
+ "direction": "利好|利空|中性",
+ "sectors": ["受影响板块", 最多3个],
+ "confidence": 0-100,
+ "reasoning": "≤150字影响传导逻辑",
+ "stocks": [{"name":"股票名","code":"6位代码","reason":"≤60字推荐理由",
+             "stars": 1到5的整数,
+             "expected_return_pct": 预计5日收益百分数(可为负),
+             "buy_zone_low": 建议购入区间下限, "buy_zone_high": 建议购入区间上限}]
+ stocks 最多 3 只；buy_zone 必须参考该股现价（快照中有 price 字段）。"""
+
+ANALYZE_COOLDOWN_S = 180
+MAX_NEWS_PER_RUN = 8
+
+
+class EventImpactService:
+
+    def __init__(self, emit, db_path):
+        self.emit = emit
+        self.db_path = str(db_path)
+        self.llm = LlmClient()
+        self.last_analyzed_at = None     # 只分析该时间之后的快讯
+        self.last_run = 0.0
+
+    def _conn(self):
+        return sqlite3.connect(self.db_path, check_same_thread=False)
+
+    def _ensure_tables(self, conn):
+        conn.execute("""CREATE TABLE IF NOT EXISTS event_impact (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            ts TEXT NOT NULL,
+            event_summary TEXT NOT NULL,
+            direction TEXT,
+            sectors TEXT,
+            confidence INTEGER,
+            reasoning TEXT,
+            stocks TEXT,
+            evidence TEXT
+        )""")
+        conn.execute("""CREATE TABLE IF NOT EXISTS news_analyzed (
+            event_time TEXT NOT NULL,
+            content TEXT NOT NULL,
+            analyzed_at TEXT NOT NULL,
+            PRIMARY KEY (event_time, content)
+        )""")
+        conn.commit()
+
+    def has_new_news(self, conn) -> bool:
+        cutoff = (datetime.now() - timedelta(hours=3)).strftime('%Y-%m-%d %H:%M:%S')
+        n = conn.execute(
+            "SELECT COUNT(*) FROM news_flash WHERE event_time >= ? AND event_time > COALESCE("
+            " (SELECT MAX(analyzed_at) FROM news_analyzed), '1970-01-01')", (cutoff,)).fetchone()[0]
+        return n > 0
+
+    def run_if_due(self):
+        """供采集循环调用：有新快讯且冷却结束才分析"""
+        import time as _t
+        if not self.llm.enabled:
+            return
+        now = _t.time()
+        if now - self.last_run < ANALYZE_COOLDOWN_S:
+            return
+        conn = self._conn()
+        try:
+            self._ensure_tables(conn)
+            if not self.has_new_news(conn):
+                return
+            self.last_run = now
+            self.analyze_latest(conn)
+        except LlmError as e:
+            self._emit('事件分析跳过', {'msg': str(e)[:120]})
+        except Exception as e:
+            traceback.print_exc()
+            self._emit('事件分析失败', {'msg': str(e)[:120]})
+        finally:
+            conn.close()
+
+    # ---------- 单次分析 ----------
+
+    def analyze_latest(self, conn):
+        news = conn.execute(
+            "SELECT event_time, content FROM news_flash "
+            "WHERE event_time >= datetime('now', '-6 hours') "
+            "ORDER BY event_time DESC LIMIT ?", (MAX_NEWS_PER_RUN,)).fetchall()
+        if not news:
+            return None
+        today = datetime.now().strftime('%Y-%m-%d')
+        top_flow = conn.execute(
+            "SELECT ts_code, ROUND(main_net_in/1e8,2) y FROM money_flow WHERE trade_date=? "
+            "ORDER BY main_net_in DESC LIMIT 8", (today,)).fetchall()
+        spot = self._market_snapshot()
+
+        evidence = [{'time': t, 'text': c[:160]} for t, c in news]
+        prompt_parts = ['## 最新快讯（原文）']
+        prompt_parts += ['[{}] {}'.format(t, c) for t, c in news]
+        prompt_parts.append('\n## 今日主力资金净流入 TOP8（亿元）')
+        prompt_parts += ['{}: +{}亿'.format(r[0], r[1]) for r in top_flow] or ['（无）']
+        prompt_parts.append('\n## 上证指数')
+        sh = conn.execute("SELECT trade_date, close FROM stock_daily WHERE ts_code='sh000001' "
+                          "ORDER BY trade_date DESC LIMIT 1").fetchone()
+        if sh:
+            prompt_parts.append('{} 收盘 {}'.format(sh[0], sh[1]))
+        prompt_parts.append('\n## 全市场个股现价快照（节选，只能从中选股）\n')
+        prompt_parts.append('代码 | 名称 | 现价 | 今日涨跌% | 主力净流入(亿)')
+        for code, price, pct, mflow in spot[:120]:
+            prompt_parts.append('{} | {:.2f} | {:.2f}% | {:.2f}亿'.format(code, price, pct, mflow))
+        user_prompt = '\n'.join(prompt_parts)
+
+        card = None
+        for attempt in range(2):
+            content = self.llm.chat(SYSTEM_PROMPT,
+                                    user_prompt + ('\n\n上一次输出不是合法 JSON，请重新输出。' if attempt else ''))
+            card = self._parse(content, spot_map=None)
+            if card:
+                break
+        if not card or not card.get('has_event'):
+            self._mark_analyzed(conn, news)
+            return None
+        self._mark_analyzed(conn, news)
+
+        card['stocks'] = self._verify_stocks(card.get('stocks') or [])
+        event = {
+            'ts': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
+            'kind': '事件影响分析',
+            'data': {
+                'event_summary': card.get('event_summary', ''),
+                'direction': card.get('direction', '中性'),
+                'sectors': card.get('sectors', []),
+                'confidence': card.get('confidence', 0),
+                'reasoning': card.get('reasoning', ''),
+                'stocks': card['stocks'],
+                'evidence': evidence,
+            },
+        }
+        self._save(conn, event)
+        self.emit(event)
+        return event
+
+    # ---------- grounding ----------
+
+    def _market_snapshot(self):
+        """全市场现价快照（新浪源，一次请求）：[(code, price, pct, main_net_in亿)]"""
+        import akshare as ak
+        df = ak.stock_zh_a_spot()
+        df.columns = [str(c) for c in df.columns]
+        out = []
+        flow = self._today_flow_map()
+        for _, r in df.iterrows():
+            raw = str(r.get('代码', ''))
+            code = raw[-6:] if raw else ''
+            if code[:2] not in ('60', '00', '30', '68'):   # 仅沪深A股
+                continue
+            try:
+                price = float(r.get('最新价'))
+            except (TypeError, ValueError):
+                continue
+            if price <= 0:
+                continue
+            out.append((code, price, float(r.get('涨跌幅') or 0), flow.get(code, 0.0)))
+        return out
+
+    def _today_flow_map(self):
+        today = datetime.now().strftime('%Y-%m-%d')
+        try:
+            return {code: round(v / 1e8, 2) for code, v in self._conn().execute(
+                "SELECT ts_code, main_net_in FROM money_flow WHERE trade_date=? AND main_net_in IS NOT NULL",
+                (today,))}
+        except sqlite3.Error:
+            return {}
+
+    def _verify_stocks(self, stocks):
+        """校验 LLM 选出的股票必须存在于真实快照；购入区间以真实现价重算"""
+        spot = {code: (price, pct, flow) for code, price, pct, flow in self._market_snapshot()}
+        verified = []
+        for s in stocks[:5]:
+            code = str(s.get('code', '')).zfill(6)[:6]
+            if code not in spot:
+                continue
+            price, pct, flow = spot[code]
+            try:
+                lo = float(s.get('buy_zone_low', price * 0.99))
+                hi = float(s.get('buy_zone_high', price * 1.01))
+            except (TypeError, ValueError):
+                lo, hi = price * 0.99, price * 1.01
+            lo, hi = min(lo, hi), max(lo, hi)
+            try:
+                er = max(-15.0, min(15.0, float(s.get('expected_return_pct', 0))))
+            except (TypeError, ValueError):
+                er = 0.0
+            try:
+                stars = max(1, min(5, int(s.get('stars', 3))))
+            except (TypeError, ValueError):
+                stars = 3
+            verified.append({
+                'name': s.get('name', ''), 'code': code,
+                'price': round(price, 2), 'pct_today': round(pct, 2),
+                'main_net_in_yi': round(flow, 2),
+                'buy_zone': [round(lo, 2), round(hi, 2)],
+                'expected_return_pct': er, 'stars': stars,
+                'reason': s.get('reason', ''),
+            })
+        verified.sort(key=lambda x: -x['stars'])
+        return verified
+
+    def _save(self, conn, event):
+        conn.execute(
+            "INSERT INTO event_impact(ts,event_summary,direction,sectors,confidence,"
+            "reasoning,stocks,evidence) VALUES (?,?,?,?,?,?,?,?)",
+            (event['ts'], event['data']['event_summary'], event['data']['direction'],
+             json.dumps(event['data'].get('sectors', []), ensure_ascii=False),
+             event['data']['confidence'], event['data']['reasoning'],
+             json.dumps(event['data']['stocks'], ensure_ascii=False),
+             json.dumps(event['data'].get('evidence', []), ensure_ascii=False)))
+        conn.commit()
+
+    def _mark_analyzed(self, conn, news):
+        now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        conn.executemany("INSERT OR REPLACE INTO news_analyzed(event_time,content,analyzed_at) "
+                         "VALUES (?,?,?)", [(t, c, now) for t, c in news])
+        conn.commit()
+
+    def _parse(self, content, spot_map=None):
+        import re
+        m = re.search(r'\{.*\}', content, re.S)
+        if not m:
+            return None
+        try:
+            return json.loads(m.group(0))
+        except json.JSONDecodeError:
+            return None
+
+    def history(self, limit=20):
+        conn = self._conn()
+        rows = conn.execute("SELECT ts,event_summary,direction,sectors,confidence,stocks,evidence "
+                            "FROM event_impact ORDER BY id DESC LIMIT ?", (limit,)).fetchall()
+        conn.close()
+        out = []
+        for ts, summary, direction, sectors, conf, stocks, evidence in rows:
+            out.append({'ts': ts, 'kind': '事件影响分析',
+                        'data': {'event_summary': summary, 'direction': direction,
+                                 'sectors': json.loads(sectors or '[]'),
+                                 'stocks': json.loads(stocks or '[]'),
+                                 'evidence': json.loads(evidence or '[]'),
+                                 'confidence': conf}})
+        return out

src/analysis/llm_client.py

@@ -0,0 +1,76 @@
+# -*- coding: utf-8 -*-
+"""
+LLM 客户端（OpenAI 兼容 /chat/completions）。
+- base_url / model / key 由环境变量配置（默认智谱 GLM）
+- 出站安全：仅 http/https、显式拒绝 localhost、解析 IP 拒绝环回/私有/保留段、
+  禁用重定向（防 DNS rebinding 绕过），值全部走 JSON 序列化，密钥不落日志
+"""
+import ipaddress
+import json
+import os
+from urllib.parse import urlparse
+
+import requests
+
+
+class LlmError(Exception):
+    pass
+
+
+def _validated_url(base_url: str) -> str:
+    u = urlparse(base_url)
+    if u.scheme not in ('http', 'https'):
+        raise LlmError('LLM base_url 仅允许 http/https')
+    host = u.hostname or ''
+    if not host or host.lower() in ('localhost', 'localhost.localdomain'):
+        raise LlmError('LLM base_url 拒绝 localhost')
+    port = u.port or (443 if u.scheme == 'https' else 80)
+    try:
+        infos = socket.getaddrinfo(host, port)
+    except socket.gaierror as e:
+        raise LlmError('LLM base_url 域名解析失败: {}'.format(host))
+    for info in infos:
+        ip = ipaddress.ip_address(info[4][0])
+        if (ip.is_loopback or ip.is_private or ip.is_link_local or ip.is_reserved
+                or ip.is_multicast or ip.is_unspecified):
+            raise LlmError('LLM base_url 拒绝非公网地址: {}'.format(ip))
+    return '{}://{}{}'.format(u.scheme, u.netloc, u.path)
+
+
+class LlmClient:
+
+    def __init__(self):
+        self.api_key = os.environ.get('JQUANT_LLM_API_KEY', '').strip()
+        self.base_url = (os.environ.get('JQUANT_LLM_BASE_URL', '').strip()
+                         or 'https://open.bigmodel.cn/api/paas/v4')
+        self.model = os.environ.get('JQUANT_LLM_MODEL', '').strip() or 'glm-4-flash'
+        self.temperature = float(os.environ.get('JQUANT_LLM_TEMPERATURE', '0.2'))
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self.api_key)
+
+    def chat(self, system_prompt: str, user_prompt: str) -> str:
+        if not self.enabled:
+            raise LlmError('未配置 LLM API Key（JQUANT_LLM_API_KEY），请在服务环境变量中设置')
+        url = _validated_url(self.base_url.rstrip('/')) + '/chat/completions'
+        body = {
+            'model': self.model,
+            'temperature': self.temperature,
+            'messages': [
+                {'role': 'system', 'content': system_prompt},
+                {'role': 'user', 'content': user_prompt},
+            ],
+        }
+        # 校验与请求紧邻；禁重定向防 DNS rebinding 绕过 IP 校验
+        r = requests.post(url, json=body, timeout=90, allow_redirects=False,
+                          headers={'Authorization': 'Bearer ' + self.api_key,
+                                   'Content-Type': 'application/json'})
+        if r.status_code in (301, 302, 303, 307, 308):
+            raise LlmError('LLM 端点发生重定向，已拒绝（防 SSRF 绕过）')
+        if r.status_code != 200:
+            raise LlmError('LLM HTTP {}: {}'.format(r.status_code, r.text[:200]))
+        content = r.json().get('choices', [{}])[0].get('message', {}).get('content')
+        if not content:
+            raise LlmError('LLM 响应缺少 content')
+        return content

src/analysis/mine_patterns.py

@@ -12,8 +12,9 @@ import pandas as pd
 
 warnings.filterwarnings('ignore')
 
-DB = r'C:' + chr(92) + 'Users' + chr(92) + 'lookt' + chr(92) + 'a_stock_timeline' + chr(92) + 'data' + chr(92) + 'a_stock.db'
+import os
-OUT = r'E:' + chr(92) + 'Data' + chr(92) + 'skills' + chr(92) + 'a-stock-timeline-patterns' + chr(92) + 'references'
+DB = os.environ.get('MINE_DB', 'data/a_stock.db')
+OUT = os.environ.get('MINE_OUT_DIR', '../skills-out/references')
 
 IDX_NAMES = {'sh000001': '上证指数', 'sz399001': '深证成指', 'sh000300': '沪深300',
              'sz399006': '创业板指', 'sh000688': '科创50'}

src/fetcher/calendar_fetcher.py

@@ -1,13 +1,15 @@
 """交易日历采集 - AkShare tool_trade_date_hist_sina"""
 import warnings, logging, sqlite3
 from datetime import datetime
+from pathlib import Path
 import pandas as pd
+from src.storage.db import get_conn
 import akshare as ak
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
 log = logging.getLogger(__name__)
 
-DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db'
+DB = str(Path(__file__).resolve().parent.parent.parent / 'data' / 'a_stock.db')
 
 def fetch_calendar():
     df = ak.tool_trade_date_hist_sina()
@@ -17,7 +19,7 @@ def fetch_calendar():
     df['trade_date'] = pd.to_datetime(df['trade_date'])
     df['is_trade_day'] = df['trade_date'].dt.dayofweek.apply(lambda x: 0 if x >= 5 else 1)
     df['trade_date'] = df['trade_date'].dt.strftime('%Y-%m-%d')
-    conn = sqlite3.connect(DB)
+    conn = get_conn()
     df.to_sql('trade_calendar', conn, if_exists='replace', index=False)
     conn.close()
     log.info(f'写入 trade_calendar: {len(df)} 行')

src/fetcher/cn_news_fetcher.py

@@ -1,13 +1,15 @@
 """A股新闻采集 - AkShare stock_news_em"""
 import warnings, logging, sqlite3
 from datetime import datetime
+from pathlib import Path
 import pandas as pd
+from src.storage.db import get_conn
 import akshare as ak
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
 log = logging.getLogger(__name__)
 
-DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db'
+DB = str(Path(__file__).resolve().parent.parent.parent / 'data' / 'a_stock.db')
 
 def fetch_news():
     df = ak.stock_news_em(symbol='A股')
@@ -32,7 +34,7 @@ def fetch_news():
     df['inserted_at'] = datetime.now().isoformat()
     if 'id' not in df.columns:
         df.insert(0, 'id', range(1, len(df) + 1))
-    conn = sqlite3.connect(DB)
+    conn = get_conn()
     df.to_sql('news_cn', conn, if_exists='replace', index=False)
     conn.close()
     log.info(f'写入 news_cn: {len(df)} 行')

src/fetcher/fund_flow_fetcher.py

@@ -4,6 +4,7 @@ import warnings
 import logging
 from urllib.parse import urlparse
 
+from pathlib import Path
 import pandas as pd
 import requests
 
@@ -13,7 +14,7 @@ warnings.filterwarnings('ignore')
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
 log = logging.getLogger(__name__)
 
-DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db'
+DB = str(Path(__file__).resolve().parent.parent.parent / 'data' / 'a_stock.db')
 
 # 出站请求域名白名单（SSRF 防护：仅 http + 白名单主机，禁重定向跟随）
 _ALLOWED_HOSTS = {'data.10jqka.com.cn'}

src/fetcher/global_index_fetcher.py

@@ -1,13 +1,15 @@
 """外盘指数采集 - AkShare index_us_stock_sina"""
 import warnings, logging, sqlite3
 from datetime import datetime
+from pathlib import Path
 import pandas as pd
+from src.storage.db import get_conn
 import akshare as ak
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
 log = logging.getLogger(__name__)
 
-DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db'
+DB = str(Path(__file__).resolve().parent.parent.parent / 'data' / 'a_stock.db')
 
 # 要采集的外盘指数
 INDICES = [
@@ -17,7 +19,7 @@ INDICES = [
 ]
 
 def fetch_global_index():
-    conn = sqlite3.connect(DB)
+    conn = get_conn()
     results = []
     for code, symbol, name in INDICES:
         try:

src/fetcher/kline_fetcher.py

@@ -0,0 +1,147 @@
+# -*- coding: utf-8 -*-
+"""
+K线获取（迁移自 JQuant 的 TDX K线能力，Python 侧改用 akshare 源）：
+- 日线（前复权）：ak.stock_zh_a_hist
+- 分钟线（1/5/15/30/60 分钟）：ak.stock_zh_a_hist_min_em
+统一入库 kline 表（code, tf, bar_time 主键，增量覆盖），供量化模型使用。
+"""
+import warnings
+from datetime import datetime, timedelta
+
+import pandas as pd
+
+warnings.filterwarnings('ignore')
+
+TF_CONFIG = {
+    'day': {'ak_period': 'daily', 'ak_fn': 'hist', 'keep_days': 400, 'per_day': 1},
+    '60': {'ak_period': '60', 'ak_fn': 'min_em', 'keep_days': 60, 'per_day': 4},
+    '30': {'ak_period': '30', 'ak_fn': 'min_em', 'keep_days': 30, 'per_day': 8},
+    '5': {'ak_period': '5', 'ak_fn': 'min_em', 'keep_days': 10, 'per_day': 48},
+    '1': {'ak_period': '1', 'ak_fn': 'min_em', 'keep_days': 2, 'per_day': 240},
+}
+
+
+class KlineFetcher:
+
+    def __init__(self, db_path):
+        self.db_path = str(db_path)
+
+    def _conn(self):
+        import sqlite3
+        conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        return conn
+
+    def ensure_table(self, conn):
+        conn.execute("""CREATE TABLE IF NOT EXISTS kline (
+            code     TEXT    NOT NULL,
+            tf       TEXT    NOT NULL,
+            bar_time TEXT    NOT NULL,
+            open REAL, high REAL, low REAL, close REAL,
+            volume REAL, amount REAL,
+            PRIMARY KEY (code, tf, bar_time))""")
+
+    def fetch(self, code: str, tf: str = 'day', days: int = None) -> pd.DataFrame:
+        """拉取单股K线并归一化列：[bar_time, open, high, low, close, volume, amount]"""
+        import akshare as ak
+        cfg = TF_CONFIG[tf]
+        days = days or cfg['keep_days']
+        end = datetime.now().strftime('%Y%m%d')
+        start = (datetime.now() - timedelta(days=days + 5)).strftime('%Y%m%d')
+        if cfg['ak_fn'] == 'hist':
+            try:
+                df = ak.stock_zh_a_hist(symbol=code, period='daily',
+                                        start_date=start, end_date=end, adjust='qfq')
+            except Exception as e:
+                # EM 被限速/拦截时回退新浪日线（前复权）
+                sym = ('sh' if code[:1] == '6' else 'sz') + code
+                try:
+                    df = ak.stock_zh_a_daily(symbol=sym, start_date=start, end_date=end,
+                                             adjust='qfq')
+                except Exception as e2:
+                    raise e2 from e
+        else:
+            df = ak.stock_zh_a_hist_min_em(symbol=code, period=cfg['ak_period'],
+                                           start_date=start + ' 09:30:00',
+                                           end_date=end + ' 15:00:00')
+        if df is None or df.empty:
+            return pd.DataFrame()
+        df.columns = [str(c).lower() for c in df.columns]
+        cmap = {}
+        for c in df.columns:
+            if c in ('时间', 'bar_time', 'date', '日期'):
+                cmap[c] = 'bar_time'
+            elif c in ('开盘', 'open'):
+                cmap[c] = 'open'
+            elif c in ('最高', 'high'):
+                cmap[c] = 'high'
+            elif c in ('最低', 'low'):
+                cmap[c] = 'low'
+            elif c in ('收盘', 'close'):
+                cmap[c] = 'close'
+            elif c in ('成交量', 'volume', 'vol'):
+                cmap[c] = 'volume'
+            elif c in ('成交额', 'amount', 'amt'):
+                cmap[c] = 'amount'
+        df = df.rename(columns=cmap)
+        keep = [c for c in ('bar_time', 'open', 'high', 'low', 'close', 'volume', 'amount')
+                if c in df.columns]
+        df = df[keep].copy()
+        df['bar_time'] = pd.to_datetime(df['bar_time'], errors='coerce').dt.strftime(
+            '%Y-%m-%d %H:%M:%S' if tf != 'day' else '%Y-%m-%d')
+        df = df.dropna(subset=['close'])
+        for c in ('open', 'high', 'low', 'close', 'volume', 'amount'):
+            if c in df.columns:
+                df[c] = pd.to_numeric(df[c], errors='coerce')
+        df = df.dropna(subset=['open'])
+        cutoff = (datetime.now() - timedelta(days=days)).strftime(
+            '%Y-%m-%d %H:%M:%S' if tf != 'day' else '%Y-%m-%d')
+        df = df[df['bar_time'] >= cutoff]
+        return df
+
+    def save(self, code: str, tf: str, df: pd.DataFrame) -> int:
+        if df is None or df.empty:
+            return 0
+        conn = self._conn()
+        try:
+            self.ensure_table(conn)
+            rows = [(code, tf, r.bar_time, r.open, r.high, r.low, r.close,
+                     getattr(r, 'volume', None), getattr(r, 'amount', None))
+                    for r in df.itertuples(index=False)]
+            conn.executemany(
+                "INSERT OR REPLACE INTO kline(code,tf,bar_time,open,high,low,close,volume,amount) "
+                "VALUES (?,?,?,?,?,?,?,?,?)", rows)
+            conn.commit()
+            return len(rows)
+        finally:
+            conn.close()
+
+    def sync(self, code: str, tf: str = 'day', days: int = None) -> int:
+        """拉取+入库，返回入库条数"""
+        try:
+            return self.save(code, tf, self.fetch(code, tf, days))
+        except Exception as e:
+            print('[kline] {} {} 失败: {}'.format(code, tf, e))
+            return 0
+
+    def load(self, code: str, tf: str = 'day', limit: int = 260) -> pd.DataFrame:
+        """从库中读取K线（时间升序）"""
+        conn = self._conn()
+        try:
+            self.ensure_table(conn)
+            return pd.read_sql(
+                "SELECT bar_time,open,high,low,close,volume,amount FROM kline "
+                "WHERE code=? AND tf=? ORDER BY bar_time DESC LIMIT ?",
+                conn, params=(code, tf, limit)).iloc[::-1].reset_index(drop=True)
+        finally:
+            conn.close()
+
+    def cleanup(self, tf: str):
+        cfg = TF_CONFIG[tf]
+        conn = self._conn()
+        try:
+            self.ensure_table(conn)
+            conn.execute("DELETE FROM kline WHERE tf=? AND bar_time < ?",
+                         (tf, (datetime.now() - timedelta(days=cfg['keep_days'])).strftime('%Y-%m-%d')))
+            conn.commit()
+        finally:
+            conn.close()

src/quant/engine.py

@@ -0,0 +1,352 @@
+# -*- coding: utf-8 -*-
+"""
+量化引擎编排：
+- 宇宙：全市场快照按成交额 TOP N（默认 300）
+- K线轮询刷新（后台线程，限速），覆盖 day + 30/5 分钟
+- 每轮刷新后：因子截面打分 → 推荐卡（Top 20）→ emit + 落库
+- 自适应微调：每日按 IC 反聩调整因子权重；推荐事后回填实现收益再反哺
+"""
+import json
+import os
+import threading
+import time
+import traceback
+from datetime import datetime
+
+import numpy as np
+import pandas as pd
+
+from src.fetcher.kline_fetcher import KlineFetcher
+from src.quant.model import (FACTOR_NAMES, WeightStore, compute_factors,
+                             cross_section_score, factor_ic_series)
+
+UNIVERSE_N = int(os.environ.get('QUANT_UNIVERSE_N', '300'))
+SCORING_INTERVAL_S = int(os.environ.get('QUANT_SCORING_INTERVAL_S', '1800'))
+KLINE_QPS = 3   # 每秒最多拉几只（限速防封）
+
+
+class QuantEngine:
+
+    def __init__(self, emit, db_path):
+        self.emit = emit
+        self.db_path = str(db_path)
+        self.fetcher = KlineFetcher(self.db_path)
+        self.weights_store = WeightStore(self.db_path)
+        self.state_lock = threading.Lock()
+        self.last_ranking = []
+        self.last_scored_at = None
+        self._stop = threading.Event()
+        self._threads = []
+
+    # ---------- 宇宙与K线刷新 ----------
+
+    def universe(self, n=UNIVERSE_N):
+        import akshare as ak
+        # 1) 新浪全市场快照
+        try:
+            df = ak.stock_zh_a_spot()
+            df['amt'] = pd.to_numeric(df.get('成交额'), errors='coerce')
+            df['code'] = df.get('代码').astype(str).str[-6:]
+            df = df[df['code'].str[:2].isin(('60', '00', '30', '68'))]
+            df = df.sort_values('amt', ascending=False).head(n)
+            pairs = list(zip(df['code'], df['名称'].astype(str)))
+            if pairs:
+                return [c for c, _ in pairs], dict(pairs)
+        except Exception as e:
+            print('[quant] 新浪宇宙失败，改用东财:', e, flush=True)
+        # 2) 东财全市场快照兜底
+        try:
+            df = ak.stock_zh_a_spot_em()
+            df['amt'] = pd.to_numeric(df.get('成交额'), errors='coerce')
+            df['code'] = df.get('代码').astype(str).str[-6:]
+            df = df[df['code'].str[:2].isin(('60', '00', '30', '68'))]
+            df = df.sort_values('amt', ascending=False).head(n)
+            pairs = list(zip(df['code'], df['名称'].astype(str)))
+            if pairs:
+                return [c for c, _ in pairs], dict(pairs)
+        except Exception as e:
+            print('[quant] 东财宇宙失败:', e, flush=True)
+        # 3) 兜底：资金流表中的活跃股
+        conn = self.fetcher._conn()
+        try:
+            rows = conn.execute("SELECT DISTINCT ts_code FROM money_flow LIMIT ?",
+                                (n,)).fetchall()
+            return [r[0] for r in rows], {}
+        finally:
+            pass
+
+    def _kline_count(self, tf):
+        import sqlite3
+        try:
+            conn = sqlite3.connect(self.db_path, check_same_thread=False)
+            n = conn.execute("SELECT COUNT(*) FROM kline WHERE tf=? AND bar_time >= datetime('now','-2 days')",
+                             (tf,)).fetchone()[0]
+            conn.close()
+            return n
+        except Exception:
+            return 0
+
+    def _kline_worker(self):
+        """后台轮询：持续刷新宇宙内 K 线（day 全量 + 30/5 分钟）"""
+        while not self._stop.is_set():
+            try:
+                codes, names = self.universe()
+                if not codes:
+                    time.sleep(120)
+                    continue
+                with self.state_lock:
+                    self.names = names
+                fails = 0
+                for code in codes:
+                    if self._stop.is_set():
+                        return
+                    for tf in ('day', '30'):
+                        before = self._kline_count(tf)
+                        self.fetcher.sync(code, tf)
+                        if self._kline_count(tf) == before:
+                            fails += 1
+                        else:
+                            fails = max(0, fails - 1)
+                    # 限速：连续失败加退避，防触发源封禁
+                    if fails and fails % 5 == 0:
+                        time.sleep(min(30, 5 + fails))
+                    time.sleep(1.2)
+                print('[quant] 本轮K线刷新: {} 只, 连续未新增 {}'.format(len(codes), fails), flush=True)
+                with self.state_lock:
+                    self.kline_ready = True
+                print('[quant] K线刷新完成一轮: {} 只'.format(len(codes)), flush=True)
+                self.run_scoring_and_push()
+                self.maybe_fine_tune()
+            except Exception as e:
+                print('[quant] K线刷新异常:', e, flush=True)
+                time.sleep(60)
+
+    def start(self):
+        t = threading.Thread(target=self._kline_worker, daemon=True, name='quant-kline')
+        self._stop.clear()
+        t.start()
+        self._threads.append(t)
+
+    def run_scoring_and_push(self, top_n=20):
+        """打分并推送推荐卡（WS + REST 共用）"""
+        try:
+            recs = self.build_recommendations(top_n)
+            if not recs:
+                return 0
+            event = {'ts': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
+                     'kind': '量化推荐',
+                     'data': {'ts': self.last_scored_at,
+                              'model': '自适应多因子模型',
+                              'list': recs}}
+            self.emit(event)
+            print('[quant] 推荐卡已推送: {} 只'.format(len(recs)), flush=True)
+            return len(recs)
+        except Exception:
+            traceback.print_exc()
+            return 0
+
+    def maybe_fine_tune(self):
+        """每日一次：事后评估回填 + IC 微调权重"""
+        today = datetime.now().strftime('%Y-%m-%d')
+        with self.state_lock:
+            if self.state.get('ft_date') == today:
+                return
+            self.state['ft_date'] = today
+        try:
+            filled = self.evaluate_pending()
+            w, notes = self.fine_tune()
+            self.emit({'ts': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
+                       'kind': '模型微调',
+                       'data': {'filled': filled, 'notes': notes,
+                                'weights': {k: round(v, 3) for k, v in w.items()}}})
+        except Exception as e:
+            print('[quant] 微调失败:', e, flush=True)
+
+    def stop(self):
+        self._stop.set()
+
+    # ---------- 打分与推荐 ----------
+
+    def score_universe(self):
+        """对已刷新K线的股票做因子打分；返回 (ranking, factor_rows)"""
+        conn = self.fetcher._conn()
+        try:
+            rows = conn.execute("SELECT DISTINCT code FROM kline WHERE tf='day'").fetchall()
+            codes = [r[0] for r in rows]
+        finally:
+            pass
+        factor_rows, klines = {}, {}
+        for code in codes:
+            k = self.fetcher.load(code, 'day', 120)
+            if len(k) < 30:
+                continue
+            f = compute_factors(k)
+            if f:
+                factor_rows[code] = f
+                klines[code] = k
+        weights = self.weights_store.load()
+        scored = cross_section_score(factor_rows, weights)
+        return scored, factor_rows, klines, weights
+
+    def build_recommendations(self, top_n=20):
+        scored, factor_rows, klines, weights = self.score_universe()
+        if not scored:
+            return []
+        names = getattr(self, 'names', {})
+        scores = [s for _, s, _ in scored]
+        smin, smax = min(scores), max(scores)
+        spread = (smax - smin) or 1
+        # 历史基准：当前权重的全历史高分股 5 日中位收益（回测口径）
+        import numpy as np
+        hist_median = None
+        try:
+            top_q = np.quantile([s for _, s, _ in scored], 0.8)
+        except Exception:
+            top_q = None
+
+        recs = []
+        for code, score, _z in scored[:top_n]:
+            k = klines.get(code)
+            if k is None or len(k) < 6:
+                continue
+            price = float(k['close'].iloc[-1])
+            name = names.get(code, code)
+            star = 1 + int(round((score - smin) / spread * 4))   # 1..5
+            contrib = sorted(_z.items(), key=lambda kv: -abs(kv[1]))[:2]
+            reason = '、'.join('{}突出'.format(f) for f, _ in contrib)
+            recs.append({
+                'ts': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
+                'code': code, 'name': name,
+                'price': round(price, 2),
+                'score': round(score, 3),
+                'buy_low': round(price * 0.99, 2),
+                'buy_high': round(price * 1.005, 2),
+                'expected_return_pct': None,   # 由回测基准填充
+                'stars': max(1, min(5, star)),
+                'reason': reason or '多因子综合打分靠前',
+            })
+        # 预计收益：用同权重下高分股历史 5 日中位收益（无历史则置空）
+        try:
+            med = self._backtest_top_median(weights)
+            for r in recs:
+                r['expected_return_pct'] = round(med * 100, 2) if med is not None else None
+        except Exception:
+            pass
+        self.last_ranking = recs
+        self.last_scored_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        self._save_recommendations(recs)
+        return recs
+
+    def _backtest_top_median(self, weights, days=60, horizon=5):
+        """在已有K线的历史上按当前权重打分，取每日 Top20% 的 5 日中位收益"""
+        conn = self.fetcher._conn()
+        try:
+            codes = [r[0] for r in conn.execute(
+                "SELECT DISTINCT code FROM kline WHERE tf='day'").fetchall()]
+        finally:
+            pass
+        closes, frames = {}, {}
+        for code in codes:
+            k = self.fetcher.load(code, 'day', 140)
+            if len(k) >= 60:
+                k['bar_date'] = k['bar_time'].str[:10]
+                closes[code] = k.set_index('bar_date')['close']
+                frames[code] = k
+        if len(closes) < 30:
+            return None
+        all_dates = sorted(set().union(*[set(c.index) for c in closes.values()]))
+        med_list = []
+        for d in all_dates[60:-horizon]:
+            fd = {}
+            for code, k in frames.items():
+                sub = k[k['bar_date'] <= d]
+                if len(sub) >= 30:
+                    fd[code] = compute_factors(sub)
+            if len(fd) < 30:
+                continue
+            scored = cross_section_score(fd, weights)
+            topq = [c for c, s, _ in scored[:max(1, len(scored) // 5)]]
+            fwd = []
+            for c in topq:
+                s = closes.get(c)
+                if s is None:
+                    continue
+                after = s[s.index > d]
+                base = s[s.index <= d].iloc[-1]
+                if len(after) >= horizon and base:
+                    fwd.append(after.iloc[horizon - 1] / base - 1)
+            if fwd:
+                med_list.append(float(pd.Series(fwd).median()))
+        return float(pd.Series(med_list).median()) if med_list else None
+
+    def _save_recommendations(self, recs):
+        conn = self.fetcher._conn()
+        try:
+            conn.execute("""CREATE TABLE IF NOT EXISTS quant_recommendation (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                ts TEXT NOT NULL, code TEXT, name TEXT, price REAL,
+                score REAL, stars INTEGER,
+                buy_low REAL, buy_high REAL,
+                expected_return_pct REAL, reason TEXT,
+                realized_return_pct REAL,
+                model_note TEXT)""")
+            now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+            conn.executemany(
+                "INSERT INTO quant_recommendation(ts,code,name,price,score,stars,"
+                "buy_low,buy_high,expected_return_pct,reason,model_note) "
+                "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
+                [(r['ts'], r['code'], r['name'], r['price'], r['score'], r['stars'],
+                  r['buy_low'], r['buy_high'], r['expected_return_pct'], r['reason'],
+                  '自适应因子模型') for r in recs])
+            conn.commit()
+        finally:
+            conn.close()
+
+    # ---------- 自适应微调 + 事后评估 ----------
+
+    def fine_tune(self):
+        """IC 反聩微调因子权重，返回调整说明"""
+        conn = self.fetcher._conn()
+        codes = [r[0] for r in conn.execute(
+            "SELECT DISTINCT code FROM kline WHERE tf='day'").fetchall()]
+        conn.close()
+        factor_by_date, close_by_code = {}, {}
+        for code in codes:
+            k = self.fetcher.load(code, 'day', 140)
+            if len(k) < 40:
+                continue
+            k['bar_date'] = k['bar_time'].str[:10]
+            close_by_code[code] = k.set_index('bar_date')['close']
+            for d, sub in k.groupby('bar_date'):
+                if len(sub) >= 30:
+                    factor_by_date.setdefault(d, {})
+                    f = compute_factors(sub)
+                    factor_by_date[d][code] = f
+        ic_map = factor_ic_series(factor_by_date, close_by_code)
+        w, notes = self.weights_store.adjust_by_ic(ic_map)
+        return w, notes
+
+    def evaluate_pending(self, horizon=5):
+        """回填历史推荐的实际 5 日收益（事后检验），供微调与展示"""
+        conn = self.fetcher._conn()
+        try:
+            rows = conn.execute(
+                "SELECT id, code, ts, realized_return_pct FROM quant_recommendation "
+                "WHERE realized_return_pct IS NULL").fetchall()
+            filled = 0
+            for rid, code, ts, _ in rows:
+                k = self.fetcher.load(code, 'day', 30)
+                if k.empty:
+                    continue
+                k = k[k['bar_time'] > ts]
+                if len(k) < horizon:
+                    continue
+                base = float(k['open'].iloc[0])
+                ret = (float(k['close'].iloc[horizon - 1]) / base - 1) * 100
+                conn.execute("UPDATE quant_recommendation SET realized_return_pct=? WHERE id=?",
+                             (round(ret, 2), rid))
+                filled += 1
+            conn.commit()
+            return filled
+        finally:
+            conn.close()

src/quant/model.py

@@ -0,0 +1,189 @@
+# -*- coding: utf-8 -*-
+"""
+自适应量化模型：多因子打分 + IC 反聩微调 + 推荐卡生成。
+
+因子（全部由日线 K 线计算，禁前视：只用截至当日的窗口）：
+  mom_20      20 日动量
+  trend_ma20  收盘相对 MA20 偏离
+  ma_align    MA5/MA20 相对位置
+  vol_ratio   量比（当日量/20日均量）
+  rsi_inv     (50-RSI14)/50（超卖得分高）
+  macd_hist   MACD 柱/现价
+  vola_inv    负 20 日波动率（低波动加分）
+
+自适应微调：每轮评估各因子近 20 日 IC（因子值 vs 后 5 日收益的秩相关），
+IC > +0.02 权重×1.05，IC < -0.02 权重×0.95（权重限制在 [0.1, 3.0]），
+并记录调整原因。打分 = Σ w_f × 截面 zscore(f)。
+"""
+import json
+import math
+import os
+from datetime import datetime
+
+import numpy as np
+import pandas as pd
+
+FACTOR_NAMES = ['mom_20', 'trend_ma20', 'ma_align', 'vol_ratio', 'rsi_inv', 'macd_hist', 'vola_inv']
+
+DEFAULT_WEIGHTS = {f: 1.0 for f in FACTOR_NAMES}
+W_MIN, W_MAX = 0.1, 3.0
+
+
+# ── 指标计算（单只股票的日线 DataFrame，时间升序，含 open/high/low/close/volume） ──
+
+def sma(s, n):
+    return s.rolling(n).mean()
+
+
+def rsi_series(close, n=14):
+    d = close.diff()
+    gain = d.clip(lower=0).rolling(n).mean()
+    loss = (-d.clip(upper=0)).rolling(n).mean()
+    out = 100 - 100 / (1 + gain / loss.replace(0, np.nan))
+    return out.fillna(50)
+
+
+def macd_hist_series(close):
+    ema12 = close.ewm(span=12, adjust=False).mean()
+    ema26 = close.ewm(span=26, adjust=False).mean()
+    dif = ema12 - ema26
+    dea = dif.ewm(span=9, adjust=False).mean()
+    return (dif - dea) / close
+
+
+def compute_factors(df: pd.DataFrame) -> dict:
+    """返回 {因子名: 因子在最末日的值}；数据不足的因子为 NaN"""
+    if df is None or len(df) < 30:
+        return {}
+    close = df['close']
+    vol = df['volume']
+    out = {}
+    out['mom_20'] = (close.iloc[-1] / close.iloc[-21] - 1) if len(close) > 20 else np.nan
+    ma20 = sma(close, 20).iloc[-1]
+    out['trend_ma20'] = (close.iloc[-1] - ma20) / ma20 if ma20 else np.nan
+    ma5, ma20s = sma(close, 5).iloc[-1], ma20
+    out['ma_align'] = (ma5 - ma20s) / ma20s if ma20s else np.nan
+    v20 = sma(vol, 20).iloc[-1]
+    out['vol_ratio'] = vol.iloc[-1] / v20 if v20 else np.nan
+    out['rsi_inv'] = (50 - rsi_series(close).iloc[-1]) / 50
+    out['macd_hist'] = macd_hist_series(close).iloc[-1]
+    out['vola_inv'] = -(close.pct_change().rolling(20).std().iloc[-1] or np.nan)
+    return out
+
+
+# ── 截面打分 ──
+
+def cross_section_score(factor_rows: dict, weights: dict):
+    """
+    factor_rows: {code: {因子: 值}}
+    返回 [(code, score, {因子: z值})] 按分降序
+    """
+    names = [f for f in FACTOR_NAMES if f in weights]
+    codes = list(factor_rows.keys())
+    z = {c: {} for c in codes}
+    for f in names:
+        vals = pd.Series([factor_rows[c].get(f, np.nan) for c in codes], index=codes, dtype=float)
+        std = vals.std()
+        if not std or std != std:
+            z_f = pd.Series(np.nan, index=codes)
+        else:
+            z_f = (vals - vals.mean()) / std
+        for c in codes:
+            z[c][f] = z_f.get(c, np.nan)
+    scored = []
+    for c in codes:
+        total, parts = 0.0, 0
+        for f in names:
+            v = z[c].get(f)
+            if v == v:   # 非 NaN
+                total += weights.get(f, 1.0) * v
+                parts += 1
+        if parts >= 3:
+            scored.append((c, total, z[c]))
+    scored.sort(key=lambda x: -x[1])
+    return scored
+
+
+def factor_ic_series(factor_by_date: dict, close_by_code: dict, days=20, horizon=5):
+    """
+    因子近 IC 序列：factor_by_date {date: {code: value}}；close_by_code {code: Series}
+    返回 {因子: 平均IC}
+    """
+    dates = sorted(factor_by_date.keys())
+    ics = {f: [] for f in FACTOR_NAMES}
+    for d in dates[-days:]:
+        fd = factor_by_date[d]
+        if len(fd) < 20:
+            continue
+        fwd = {}
+        for code, v in fd.items():
+            s = close_by_code.get(code)
+            if s is None:
+                continue
+            after = s[s.index > d]
+            if len(after) > horizon:
+                fwd[code] = after.iloc[horizon] / s[s.index <= d].iloc[-1] - 1
+        if len(fwd) < 20:
+            continue
+        codes = list(fwd.keys())
+        for f in FACTOR_NAMES:
+            fv = pd.Series([fd[c].get(f, np.nan) for c in codes], dtype=float)
+            rv = pd.Series([fwd[c] for c in codes], dtype=float)
+            ok = fv.notna() & rv.notna()
+            if ok.sum() < 20:
+                continue
+            ics[f].append(fv[ok].corr(rv[ok], method='spearman'))
+    return {f: (float(np.nanmean(v)) if v else 0.0) for f, v in ics.items()}
+
+
+class WeightStore:
+
+    def __init__(self, db_path):
+        self.db_path = str(db_path)
+        self._ensure()
+
+    def _conn(self):
+        import sqlite3
+        conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        return conn
+
+    def _ensure(self):
+        with self._conn() as conn:
+            conn.execute("""CREATE TABLE IF NOT EXISTS quant_weights (
+                factor TEXT PRIMARY KEY,
+                weight REAL,
+                updated_at TEXT,
+                note TEXT)""")
+
+    def load(self) -> dict:
+        with self._conn() as conn:
+            rows = conn.execute("SELECT factor, weight FROM quant_weights").fetchall()
+        w = dict(DEFAULT_WEIGHTS)
+        w.update({r[0]: r[1] for r in rows})
+        return w
+
+    def save(self, weights: dict, note: str):
+        now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        with self._conn() as conn:
+            for f, w in weights.items():
+                conn.execute("INSERT OR REPLACE INTO quant_weights(factor,weight,updated_at,note) "
+                             "VALUES (?,?,?,?)", (f, round(w, 4), now, note))
+
+    def adjust_by_ic(self, ic_map: dict, lr=0.05):
+        """IC 反聩微调：IC>0.02 权重×(1+lr)，IC<-0.02 ×(1-lr)，夹在 [W_MIN, W_MAX]"""
+        w = self.load()
+        notes = []
+        for f, ic in ic_map.items():
+            old = w.get(f, 1.0)
+            if ic > 0.02:
+                w[f] = min(W_MAX, old * (1 + lr))
+                tag = '↑'
+            elif ic < -0.02:
+                w[f] = max(W_MIN, old * (1 - lr))
+                tag = '↓'
+            else:
+                continue
+            notes.append('{} {}{:.3f}→{:.3f} (IC{:+.3f})'.format(f, tag, old, w[f], ic))
+        if notes:
+            self.save(w, note='IC微调: ' + '; '.join(notes))
+        return w, notes

src/realtime/collector.py

@@ -0,0 +1,275 @@
+# -*- coding: utf-8 -*-
+"""
+时间线采集引擎（跨平台）：将 realtime_analyzer 的分析循环重构为可嵌入的采集器。
+- 盘前：美股隔夜收盘 -> 实证先验（a-stock-timeline-patterns skill）输出今日预估
+- 盘中：东财 7x24 快讯增量入库 + 关键词告警 + 上证实时点位
+- 盘后：龙虎榜入库 + 收盘日报
+事件通过回调 emit(event: dict) 交给上层（桌面版写文件，B/S 版走 WebSocket 广播）。
+"""
+import json
+import os
+import sqlite3
+import traceback
+from datetime import datetime, timedelta
+from pathlib import Path
+
+import requests
+
+ROOT = Path(__file__).resolve().parent.parent.parent
+DB = ROOT / 'data' / 'a_stock.db'
+FEED = ROOT / 'data' / 'realtime_feed.jsonl'
+SKILL_REF = Path(r'E:\Data\skills\a-stock-timeline-patterns\references\overnight_transmission.md')
+
+KEYWORDS = ['降息', '降准', '加息', '关税', '制裁', '收购', '重组', '国债', '证监会', 'PMI', 'CPI']
+
+# 出站请求域名白名单（SSRF 防护）
+_ALLOWED_HOSTS = {'np-weblist.eastmoney.com', 'np-listapi.eastmoney.com'}
+
+
+def _safe_get(url, params=None, timeout=10):
+    from urllib.parse import urlparse
+    u = urlparse(url)
+    if u.scheme != 'https' or u.hostname not in _ALLOWED_HOSTS:
+        raise ValueError('blocked non-allowlist url: %s' % url)
+    return requests.get(url, params=params, timeout=timeout,
+                        headers={'User-Agent': 'Mozilla/5.0'}, allow_redirects=False)
+
+
+class TimelineCollector:
+    """常驻采集引擎：一个线程安全的同步循环，由上层调度（线程/异步 executor）"""
+
+    def __init__(self, emit, db_path=None):
+        self.emit = emit                        # callable(dict)
+        self.db_path = str(db_path or DB)
+        self.state = {'date': datetime.now().strftime('%Y-%m-%d')}
+        FEED.parent.mkdir(parents=True, exist_ok=True)   # 事件流水目录
+        self.priors = {}
+        self._load_priors()
+        from src.analysis.event_impact import EventImpactService
+        self.impact = EventImpactService(emit=self.emit, db_path=self.db_path)
+
+    # ---------- 基础 ----------
+
+    def _conn(self):
+        conn = sqlite3.connect(self.db_path, check_same_thread=False)
+        return conn
+
+    def _emit(self, kind, data):
+        rec = {'ts': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'kind': kind, 'data': data}
+        try:
+            FEED.parent.mkdir(parents=True, exist_ok=True)
+            with open(FEED, 'a', encoding='utf-8') as f:
+                f.write(json.dumps(rec, ensure_ascii=False) + '\n')
+        except OSError:
+            pass
+        try:
+            self.emit(rec)
+        except Exception:
+            pass   # 回调异常不允许杀死采集循环
+
+    # ---------- 实证先验（skill 注入） ----------
+
+    def _load_priors(self):
+        import re
+        self.priors = {}
+        path = str(SKILL_REF)
+        if not os.path.exists(path):
+            # Linux 部署：skill 文件可放项目 docs/ 下
+            alt = Path(__file__).resolve().parent.parent.parent / 'docs' / 'overnight_transmission.md'
+            if not alt.exists():
+                return
+            path = str(alt)
+        try:
+            with open(path, encoding='utf-8') as f:
+                cur_us = cur_idx = None
+                for line in f:
+                    m = re.match(r'### (\S+)T日 -> (\S+)T\+1日', line)
+                    if m:
+                        cur_us, cur_idx = m.group(1), m.group(2)
+                        continue
+                    m = re.match(
+                        r'- 美股(\S+?)（(\d+)天）: A股跳空 N=\d+, 均值 (-?[\d.]+)%, 中位数 (-?[\d.]+)%, '
+                        r'胜率 ([\d.]+)%; 日内 N=\d+, 均值 (-?[\d.]+)%, 中位数 (-?[\d.]+)%, 胜率 ([\d.]+)%',
+                        line)
+                    if m and cur_us and cur_idx:
+                        self.priors[(cur_us, cur_idx, m.group(1))] = {
+                            'n': int(m.group(2)),
+                            'gap_mean': float(m.group(3)), 'gap_median': float(m.group(4)),
+                            'gap_win': float(m.group(5)),
+                            'intra_mean': float(m.group(6)), 'intra_median': float(m.group(7)),
+                            'intra_win': float(m.group(8)),
+                        }
+        except OSError:
+            pass
+
+    # ---------- 数据获取 ----------
+
+    def us_overnight(self, conn):
+        out = {}
+        for code in ('US.NDX', 'US.SPX', 'US.DJI'):
+            row = conn.execute(
+                "SELECT trade_date, pct_change FROM global_index WHERE index_code=? "
+                "ORDER BY trade_date DESC LIMIT 1", (code,)).fetchone()
+            if row:
+                out[code] = {'date': row[0], 'pct': row[1] or 0.0}
+        return out
+
+    def forecast_from_priors(self, us):
+        if not self.priors or 'US.NDX' not in us:
+            return None
+        pct = us['US.NDX']['pct'] / 100.0
+        if pct <= -0.015:
+            b = '跌>1.5%'
+        elif pct <= -0.005:
+            b = '跌0.5~1.5%'
+        elif pct < 0.005:
+            b = '正负0.5%内'
+        elif pct < 0.015:
+            b = '涨0.5~1.5%'
+        else:
+            b = '涨>1.5%'
+        k = ('US.NDX', 'sh000001', b)
+        if k not in self.priors:
+            return None
+        p = self.priors[k]
+        return ('隔夜预估[美股纳指{:+.2f}% -> 分箱"{}"]: 历史上上证次日跳空均值 {:+.2f}%'
+                '（低开概率 {:.0f}%），日内均值 {:+.2f}%（日内收涨概率 {:.0f}%），全天均值 {:+.2f}%。'
+                '（样本{}天，美股先验，仅参考）').format(
+            us['US.NDX']['pct'], b, p['gap_mean'], 100 - p['gap_win'],
+            p['intra_mean'], p['intra_win'], p['gap_mean'] + p['intra_mean'], p['n'])
+
+    def index_spot_sh(self):
+        try:
+            import akshare as ak
+            df = ak.stock_zh_index_spot_em(symbol='上证系列指数')
+            row = df[df['名称'] == '上证指数']
+            if not row.empty:
+                r = row.iloc[0]
+                return {'price': float(r['最新价']), 'pct': float(r['涨跌幅'])}
+        except Exception:
+            pass
+        try:
+            import akshare as ak
+            df = ak.stock_zh_index_spot_sina()
+            row = df[df['代码'] == 'sh000001']
+            if not row.empty:
+                r = row.iloc[0]
+                return {'price': float(r['最新价']), 'pct': float(r['涨跌幅'])}
+        except Exception:
+            pass
+        return None
+
+    def fetch_em_flash(self, conn):
+        """东财 7x24 快讯增量入库，返回 (time, text) 新增列表"""
+        fresh = []
+        try:
+            r = _safe_get('https://np-weblist.eastmoney.com/comm/web/getFastNewsList',
+                          params={'client': 'web', 'biz': 'web_724', 'fastColumn': '102',
+                                  'sortEnd': '', 'pageSize': '20', 'req_trace': '1'})
+            data = r.json().get('data', {}) or {}
+            for n in data.get('fastNewsList', []) or []:
+                ts = n.get('showTime', '')
+                summary = (n.get('summary') or n.get('title') or '').strip()
+                if not ts or not summary:
+                    continue
+                dup = conn.execute(
+                    "SELECT 1 FROM news_flash WHERE event_time=? AND content=? LIMIT 1",
+                    (ts, summary)).fetchone()
+                if not dup:
+                    conn.execute(
+                        "INSERT INTO news_flash(content, event_time, source, inserted_at) "
+                        "VALUES (?,?,?,datetime('now','localtime'))", (summary, ts, '东财7x24'))
+                    fresh.append((ts, summary[:200]))
+            conn.commit()
+        except Exception as e:
+            self._emit('采集错误', {'msg': str(e)[:120]})
+        return fresh
+
+    def save_lhb_today(self, conn):
+        try:
+            from src.fetcher.lhb_fetcher import fetch_lhb_date
+            from src.storage.db import upsert_rows
+            today = datetime.now().strftime('%Y-%m-%d')
+            df = fetch_lhb_date(today)
+            if df is not None and not df.empty:
+                return upsert_rows(df, 'lhb_daily',
+                                   conflict_cols=['trade_date', 'ts_code', 'reason'])
+        except Exception as e:
+            self._emit('采集错误', {'msg': str(e)[:120]})
+        return 0
+
+    def news_alerts(self, conn):
+        hits = []
+        cutoff = (datetime.now() - timedelta(minutes=30)).strftime('%Y-%m-%d %H:%M:%S')
+        for table, tcol, ccol in (('news_flash', 'event_time', 'content'),
+                                  ('news_cn', 'pub_date', 'title')):
+            try:
+                rows = conn.execute(
+                    "SELECT {}, {} FROM {} WHERE {} >= ? ORDER BY {} DESC LIMIT 50".format(
+                        tcol, ccol, table, tcol, tcol), (cutoff,)).fetchall()
+            except sqlite3.OperationalError:
+                continue
+            for ts, text in rows:
+                for kw in KEYWORDS:
+                    if kw in (text or ''):
+                        hits.append({'time': ts, 'kw': kw, 'text': (text or '')[:80]})
+                        break
+        return hits
+
+    # ---------- 单轮采集 ----------
+
+    def one_cycle(self):
+        conn = self._conn()
+        try:
+            now = datetime.now()
+            hm = now.hour * 100 + now.minute
+
+            if 700 <= hm < 925 and not self.state.get('premarket_done'):
+                us = self.us_overnight(conn)
+                fc = self.forecast_from_priors(us)
+                self._emit('盘前隔夜预估', {'us': us, 'forecast': fc})
+                self.state['premarket_done'] = True
+
+            if 925 <= hm < 1505:
+                for ts, txt in self.fetch_em_flash(conn):
+                    self._emit('新快讯', {'time': ts, 'text': txt})
+                spot = self.index_spot_sh()
+                if spot and spot.get('price', 0) > 0:
+                    self._emit('盘中点位', {'上证': spot['price'], '涨跌幅%': spot['pct']})
+                for h in self.news_alerts(conn):
+                    self._emit('快讯关键词告警', h)
+                self.impact.run_if_due()
+
+            if hm >= 1510 and not self.state.get('postmarket_done'):
+                n = self.save_lhb_today(conn)
+                sh = conn.execute("SELECT trade_date, close FROM stock_daily "
+                                  "WHERE ts_code='sh000001' ORDER BY trade_date DESC LIMIT 1").fetchone()
+                self._emit('收盘日报', {'龙虎榜新增': n, '上证最新收盘': sh})
+                self.state['postmarket_done'] = True
+
+            if self.state.get('date') != now.strftime('%Y-%m-%d'):
+                self.state.clear()
+                self.state['date'] = now.strftime('%Y-%m-%d')
+                self._load_priors()
+                self._emit('日切', {'date': self.state['date']})
+        finally:
+            conn.close()
+
+    def run_forever(self, interval=60, on_error=None):
+        """阻塞式常驻循环（Linux 服务/桌面版均可直接调用）"""
+        while True:
+            try:
+                self.one_cycle()
+            except Exception:
+                traceback.print_exc()
+                if on_error:
+                    on_error()
+            time.sleep(interval)
+
+
+if __name__ == '__main__':
+    c = TimelineCollector(emit=lambda rec: print(
+        '[{}] {} {}'.format(rec['ts'], rec['kind'],
+                            json.dumps(rec['data'], ensure_ascii=False)), flush=True))
+    print('时间线采集引擎启动（Ctrl+C 停止）', flush=True)
+    c.run_forever()

src/web/index.html

@@ -0,0 +1,206 @@
+<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+<meta charset="UTF-8">
+<meta name="viewport" content="width=device-width, initial-scale=1.0">
+<title>JQuant · A股时间线实时监控</title>
+<style>
+  :root {
+    --bg: #0b0e14; --surface: #131722; --border: #1e2634;
+    --text: #e6e9f0; --text2: #9aa4b8;
+    --up: #f5455c; --down: #2fbf71; --accent: #4c8dff;
+  }
+  * { box-sizing: border-box; }
+  body {
+    margin: 0; background: var(--bg); color: var(--text);
+    font-family: "Segoe UI", "Microsoft YaHei", sans-serif; font-size: 14px;
+  }
+  header {
+    padding: 14px 22px; border-bottom: 1px solid var(--border);
+    display: flex; justify-content: space-between; align-items: center;
+    background: var(--surface);
+  }
+  header h1 { margin: 0; font-size: 17px; }
+  header h1 span { color: var(--accent); }
+  .conn { font-size: 12px; padding: 3px 10px; border-radius: 10px; }
+  .conn.on { background: rgba(47,191,113,.15); color: var(--down); }
+  .conn.off { background: rgba(245,69,92,.15); color: var(--up); }
+  .stats { padding: 8px 22px; color: var(--text2); font-size: 12px; border-bottom: 1px solid var(--border); }
+  main { padding: 14px 22px; max-width: 1100px; margin: 0 auto; }
+  .filters { margin-bottom: 10px; }
+  .filters button {
+    background: var(--surface); color: var(--text2); border: 1px solid var(--border);
+    border-radius: 14px; padding: 4px 14px; margin-right: 6px; cursor: pointer; font-size: 12px;
+  }
+  .filters button.active { border-color: var(--accent); color: var(--accent); }
+  ul#feed { list-style: none; margin: 0; padding: 0; }
+  li.evt {
+    background: var(--surface); border: 1px solid var(--border); border-radius: 8px;
+    padding: 10px 14px; margin-bottom: 8px; display: flex; gap: 12px; align-items: baseline;
+    animation: slidein .25s ease;
+  }
+  li.evt.fresh { border-color: var(--accent); }
+  @keyframes slidein { from { transform: translateY(-6px); opacity: 0; } to { opacity: 1; } }
+  .t { color: var(--text2); font-size: 12px; min-width: 140px; font-variant-numeric: tabular-nums; }
+  .k { min-width: 110px; font-weight: 700; }
+  .k.盘中点位 { color: var(--accent); }
+  .k.新快讯 { color: #ffb74d; }
+  .k.快讯关键词告警 { color: var(--up); }
+  .k.盘前隔夜预估, .k.收盘日报 { color: var(--down); }
+  .d { flex: 1; word-break: break-all; color: #c3cad8; }
+  .empty { text-align: center; color: var(--text2); padding: 40px 0; }
+  .impact { margin-top: 8px; border-top: 1px dashed var(--border); padding-top: 8px; }
+  .impact table { width: 100%; border-collapse: collapse; font-size: 12px; }
+  .impact th { color: var(--text2); text-align: left; padding: 3px 8px; font-weight: 600; }
+  .impact td { padding: 4px 8px; border-top: 1px solid var(--border); font-variant-numeric: tabular-nums; }
+  .stars { color: #ffb74d; letter-spacing: 1px; }
+  .dir-bull { color: var(--up); font-weight: 700; }
+  .dir-bear { color: var(--down); font-weight: 700; }
+  .dir-neutral { color: var(--text2); }
+  .conf { font-size: 11px; color: var(--text2); }
+  .disclaimer { margin-top: 6px; font-size: 11px; color: #6b7280; }
+</style>
+</head>
+<body>
+<header>
+  <h1><span>JQ</span> · A股时间线实时监控 <small style="font-weight:400;font-size:12px;color:var(--text2)">B/S（Linux 服务端推送）</small></h1>
+  <span id="conn" class="conn off">未连接</span>
+</header>
+<div class="stats">服务端：抓取（TDX/东财/同花顺）+ 分析（波动分解/事件归因）每 60s 一轮 → WebSocket 实时推送本页 | 缓冲事件 <b id="buf">0</b> 条</div>
+<main>
+  <div class="filters" id="filters">
+    <button data-f="" class="active">全部</button>
+    <button data-f="盘中点位">点位</button>
+    <button data-f="新快讯">快讯</button>
+    <button data-f="快讯关键词告警">告警</button>
+    <button data-f="盘前隔夜预估">盘前预估</button>
+    <button data-f="收盘日报">日报</button>
+    <button data-f="事件影响分析">事件影响</button>
+    <button data-f="量化推荐">量化推荐</button>
+    <button data-f="模型微调">微调</button>
+  </div>
+  <ul id="feed"><li class="empty">等待服务端推送…</li></ul>
+</main>
+<section style="max-width:1100px;margin:0 auto 30px">
+  <h2 style="font-size:15px;color:var(--text)">量化模型 · 自适应推荐（多因子 + IC 微调）</h2>
+  <div style="font-size:12px;color:var(--text2);margin-bottom:8px">权重：<span id="qw">加载中…</span></div>
+  <div style="background:var(--surface);border:1px solid var(--border);border-radius:8px;overflow:auto">
+    <table style="width:100%;border-collapse:collapse;font-size:13px" id="qtable">
+      <thead><tr style="color:var(--text2);text-align:left">
+        <th style="padding:8px 10px">代码</th><th style="padding:8px 10px">名称</th>
+        <th style="padding:8px 10px">现价</th><th style="padding:8px 10px">模型分</th>
+        <th style="padding:8px 10px">购入区间</th><th style="padding:8px 10px">预计收益(回测口径)</th>
+        <th style="padding:8px 10px">推荐指数</th><th style="padding:8px 10px">简易原因</th>
+      </tr></thead><tbody></tbody>
+    </table>
+  </div>
+  <div style="margin-top:6px;font-size:11px;color:#6b7280">⚠ 购入区间与预计收益为模型回测/推测口径，不构成投资建议。</div>
+</section>
+<script>
+const feed = document.getElementById('feed')
+const conn = document.getElementById('conn')
+const buf = document.getElementById('buf')
+let filter = ''
+let ws, retryTimer
+
+function esc(s) { return String(s ?? '').replace(/[<>&]/g, '') }
+
+function impactCard(d) {
+  if (!d || !d.stocks || !d.stocks.length) return ''
+  const dirCls = { '利好': 'dir-bull', '利空': 'dir-bear' }[d.direction] || 'dir-neutral'
+  const rows = d.stocks.map(s =>
+    `<tr><td>${esc(s.name)}</td><td>${esc(s.code)}</td>` +
+    `<td>${s.buy_zone[0]} ~ ${s.buy_zone[1]}</td>` +
+    `<td class="${s.expected_return_pct > 0 ? 'up' : s.expected_return_pct < 0 ? 'down' : ''}">${s.expected_return_pct > 0 ? '+' : ''}${s.expected_return_pct}%</td>` +
+    `<td class="stars">${'★'.repeat(s.stars)}${'☆'.repeat(5 - s.stars)}</td>` +
+    `<td>${esc(s.reason)}</td></tr>`).join('')
+  return `<div class="impact"><table>` +
+    `<tr><th>标的</th><th>代码</th><th>建议购入区间</th><th>预计收益(5日,推测)</th><th>推荐指数</th><th>简易原因</th></tr>${rows}</table>` +
+    `<div class="disclaimer">⚠ 模型推断仅供参考，预计收益为推测值，不构成投资建议；置信度 ${d.confidence ?? '-'}%</div></div>`
+}
+
+function render(e, fresh) {
+  if (filter && e.kind !== filter) return
+  const empty = feed.querySelector('.empty')
+  if (empty) empty.remove()
+  const li = document.createElement('li')
+  li.className = 'evt' + (fresh ? ' fresh' : '')
+  const d = e.data || {}
+  if (e.kind === '事件影响分析') {
+    const dirCls = { '利好': 'dir-bull', '利空': 'dir-bear', '中性': 'dir-neutral' }[d.direction] || ''
+    const sectors = (d.sectors || []).map(s => `<span class="stars">${esc(s)}</span>`).join(' / ')
+    li.innerHTML = `<span class="t">${e.ts}</span>` +
+      `<span class="k ${e.kind}">事件影响</span>` +
+      `<span class="d"><b>${esc(d.event_summary)}</b><br>` +
+      `方向：<span class="${dirCls}">${esc(d.direction)}</span> ｜ 板块：${sectors || '—'} ｜ 置信度 ${d.confidence ?? '-'}%<br>` +
+      `${esc(d.reasoning || '')}${impactCard(d)}</span>`
+  } else if (e.kind === '量化推荐') {
+    renderQuant(e.data)
+    const txt = '自适应模型推送 ' + ((d.list || []).length) + ' 只推荐'
+    li.innerHTML = `<span class="t">${e.ts}</span><span class="k ${e.kind}">${e.kind}</span><span class="d">${esc(txt)}</span>`
+  } else {
+    const txt = typeof d === 'object' ? JSON.stringify(d) : String(d)
+    li.innerHTML = `<span class="t">${e.ts}</span><span class="k ${e.kind}">${e.kind}</span><span class="d">${esc(txt)}</span>`
+  }
+  feed.prepend(li)
+  while (feed.children.length > 300) feed.lastChild.remove()
+}
+
+function applyRecent(events) {
+  [...events].reverse().forEach(e => render(e, false))
+}
+
+function function renderQuant(d) {
+  const tb = document.querySelector('#qtable tbody')
+  if (!tb) return
+  tb.innerHTML = (d.list || []).map(r =>
+    '<tr><td style="padding:6px 10px">' + esc(r.code) + '</td>' +
+    '<td style="padding:6px 10px">' + esc(r.name) + '</td>' +
+    '<td style="padding:6px 10px">' + r.price + '</td>' +
+    '<td style="padding:6px 10px">' + r.score + '</td>' +
+    '<td style="padding:6px 10px">' + r.buy_low + ' ~ ' + r.buy_high + '</td>' +
+    '<td style="padding:6px 10px">' + (r.expected_return_pct == null ? '—' : r.expected_return_pct + '%') + '</td>' +
+    '<td style="padding:6px 10px" class="stars">' + '★'.repeat(r.stars) + '</td>' +
+    '<td style="padding:6px 10px">' + esc(r.reason) + '</td></tr>').join('')
+}
+function loadQuant() {
+  fetch('/api/quant/recommendations').then(r => r.json()).then(renderQuant).catch(() => {})
+  fetch('/api/quant/weights').then(r => r.json()).then(w => {
+    document.getElementById('qw').textContent =
+      Object.entries(w || {}).map(([k, v]) => k + '=' + Number(v).toFixed(2)).join('  ') || '—'
+  }).catch(() => {})
+}
+loadQuant()
+setInterval(loadQuant, 60000)
+connect() {
+  const proto = location.protocol === 'https:' ? 'wss' : 'ws'
+  ws = new WebSocket(`${proto}://${location.host}/ws`)
+  ws.onopen = () => { conn.textContent = '已连接'; conn.className = 'conn on' }
+  ws.onclose = () => {
+    conn.textContent = '已断开，重连中…'; conn.className = 'conn off'
+    retryTimer = setTimeout(connect, 3000)
+  }
+  ws.onmessage = m => {
+    try { render(JSON.parse(m.data), true) } catch (e) {}
+  }
+}
+
+document.getElementById('filters').addEventListener('click', ev => {
+  if (ev.target.dataset.f === undefined) return
+  filter = ev.target.dataset.f
+  document.querySelectorAll('.filters button').forEach(b => b.classList.toggle('active', b === ev.target))
+  fetch(`/api/recent`).then(r => r.json()).then(d => {
+    feed.innerHTML = ''
+    ;[...(d.events || [])].reverse().forEach(e => render(e, false))
+  })
+})
+
+fetch('/api/stats').then(r => r.json()).then(s => buf.textContent = s.buffered)
+fetch('/api/recent').then(r => r.json()).then(d => {
+  ;[...(d.events || [])].reverse().forEach(e => render(e, false))
+})
+setInterval(() => fetch('/api/stats').then(r => r.json()).then(s => buf.textContent = s.buffered), 10000)
+connect()
+</script>
+</body>
+</html>

src/web/server.py

@@ -0,0 +1,158 @@
+# -*- coding: utf-8 -*-
+"""
+B/S 服务端：时间线采集引擎（后台线程）+ WebSocket 实时推送 + Web 仪表盘
+Linux 服务器常驻运行：  python -m src.web.server   （或 systemd，见 deploy/a-stock-timeline.service）
+端口默认 8100，可用环境变量 PORT 覆盖。
+"""
+import asyncio
+import json
+import os
+import threading
+import time
+from collections import deque
+from pathlib import Path
+
+from aiohttp import WSMsgType, web
+
+import sys
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
+
+from src.realtime.collector import TimelineCollector   # noqa: E402
+from src.analysis.event_impact import EventImpactService   # noqa: E402
+from src.quant.engine import QuantEngine   # noqa: E402
+
+PORT = int(os.environ.get('PORT', '8100'))
+ROOT = Path(__file__).resolve().parent.parent.parent
+WEB_DIR = Path(__file__).resolve().parent
+RECENT_MAX = 500
+
+recent = deque(maxlen=RECENT_MAX)          # 最近事件（内存）
+clients = set()                            # 活跃 WS 连接
+
+
+async def broadcast(event: dict):
+    recent.append(event)
+    payload = json.dumps(event, ensure_ascii=False)
+    dead = set()
+    for ws in clients:
+        try:
+            await ws.send_str(payload)
+        except Exception:
+            dead.add(ws)
+    for ws in dead:
+        clients.discard(ws)
+
+
+async def collector_task(_app):
+    """把阻塞式采集循环放进线程池，事件桥接到 asyncio 广播"""
+    loop = asyncio.get_running_loop()
+    collector = TimelineCollector(emit=lambda rec: loop.call_soon_threadsafe(
+        asyncio.ensure_future, broadcast(rec)))
+
+    async def poll():
+        while True:
+            try:
+                await loop.run_in_executor(None, collector.one_cycle)
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                import traceback
+                traceback.print_exc()
+            await asyncio.sleep(60)
+
+    task = asyncio.create_task(poll())
+
+    # 量化引擎：K线刷新 + 自适应打分 + 推荐卡推送
+    quant = QuantEngine(emit=lambda rec: loop.call_soon_threadsafe(
+        asyncio.ensure_future, broadcast(rec)), db_path=str(ROOT / 'data' / 'a_stock.db'))
+    _app['quant'] = quant
+    quant.start()
+
+    def _quant_loops():
+        while True:
+            try:
+                time.sleep(SCORING_INTERVAL_S)
+                quant.run_scoring_and_push()
+            except Exception as e:
+                print('[quant-loop]', e, flush=True)
+                time.sleep(60)
+
+    threading.Thread(target=_quant_loops, daemon=True, name='quant-scoring').start()
+
+    task = asyncio.create_task(poll())
+    # 启动即推最近历史（从 jsonl 恢复）
+    feed = ROOT / 'data' / 'realtime_feed.jsonl'
+    if feed.exists():
+        try:
+            lines = feed.read_text(encoding='utf-8').strip().splitlines()[-200:]
+            for line in lines:
+                try:
+                    recent.append(json.loads(line))
+                except json.JSONDecodeError:
+                    pass
+        except OSError:
+            pass
+    yield
+    task.cancel()
+
+
+async def index(_request):
+    return web.FileResponse(WEB_DIR / 'index.html')
+
+
+async def recent_events(_request):
+    return web.json_response({'events': list(recent)})
+
+
+async def quant_recommendations(_request):
+    q = _request.app['quant']
+    return web.json_response({'ts': q.last_scored_at,
+                              'list': (q.last_ranking or [])[:50]})
+
+
+async def quant_weights(_request):
+    q = _request.app['quant']
+    return web.json_response(q.weights_store.load())
+
+
+async def impact_history(_request):
+    collector = _request.app['impact']
+    return web.json_response({'events': collector.history(20)})
+
+
+async def stats(_request):
+    return web.json_response({'clients': len(clients), 'buffered': len(recent)})
+
+
+async def ws_handler(request):
+    ws = web.WebSocketResponse(heartbeat=30)
+    await ws.prepare(request)
+    clients.add(ws)
+    await ws.send_str(json.dumps({'kind': '连接成功',
+                                  'data': {'msg': '实时推送已连接', 'buffered': len(recent)}},
+                                 ensure_ascii=False))
+    try:
+        async for msg in ws:
+            if msg.type == WSMsgType.ERROR:
+                break
+    finally:
+        clients.discard(ws)
+    return ws
+
+
+def build_app():
+    app = web.Application()
+    app.router.add_get('/', index)
+    app.router.add_get('/api/recent', recent_events)
+    app.router.add_get('/api/stats', stats)
+    app.router.add_get('/api/impact', impact_history)
+    app.router.add_get('/api/quant/recommendations', quant_recommendations)
+    app.router.add_get('/api/quant/weights', quant_weights)
+    app.router.add_get('/ws', ws_handler)
+    app.cleanup_ctx.append(collector_task)
+    return app
+
+
+if __name__ == '__main__':
+    print('JQuant 时间线 B/S 服务启动: http://0.0.0.0:{} （WS: /ws）'.format(PORT))
+    web.run_app(build_app(), host='0.0.0.0', port=PORT)

start_server.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+cd "$(dirname "$0")"
+exec python3 -m src.web.server