linux-web: B/S 架构——采集引擎后台线程 + aiohttp WebSocket 实时推送 + Web 仪表盘 + systemd 部署

src/realtime/collector.py

@@ -0,0 +1,267 @@
+# -*- 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')}
+        self.priors = {}
+        self._load_priors()
+
+    # ---------- 基础 ----------
+
+    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
+        self.emit(rec)
+
+    # ---------- 实证先验（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[:60]))
+            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
+            from src.storage.db import upsert_rows
+            df = fetch_lhb()
+            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)
+
+            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,125 @@
+<!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; }
+</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>
+  </div>
+  <ul id="feed"><li class="empty">等待服务端推送…</li></ul>
+</main>
+<script>
+const feed = document.getElementById('feed')
+const conn = document.getElementById('conn')
+const buf = document.getElementById('buf')
+let filter = ''
+let ws, retryTimer
+
+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 = typeof e.data === 'object' ? JSON.stringify(e.data) : (e.data ?? '')
+  li.innerHTML = `<span class="t">${e.ts}</span><span class="k ${e.kind}">${e.kind}</span><span class="d">${d.replace(/[<>&]/g, '')}</span>`
+  feed.prepend(li)
+  while (feed.children.length > 300) feed.lastChild.remove()
+}
+
+function applyRecent(events) {
+  [...events].reverse().forEach(e => render(e, false))
+}
+
+function 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,110 @@
+# -*- 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
+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
+
+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:
+            await loop.run_in_executor(None, collector.one_cycle)
+            await asyncio.sleep(60)
+
+    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 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('/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)