182 lines
6.1 KiB
Python
182 lines
6.1 KiB
Python
# -*- 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'))
|
||
SCORING_INTERVAL_S = int(os.environ.get('QUANT_SCORING_INTERVAL_S', '1800'))
|
||
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():
|
||
time.sleep(5) # 启动即先跑一轮打分（K线可能不足，引擎会自行跳过）
|
||
while True:
|
||
try:
|
||
quant.run_scoring_and_push()
|
||
except Exception as e:
|
||
print('[quant-loop]', e, flush=True)
|
||
time.sleep(SCORING_INTERVAL_S) # 每轮之间强制间隔
|
||
|
||
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']
|
||
if q.last_ranking:
|
||
return web.json_response({'ts': q.last_scored_at,
|
||
'list': (q.last_ranking or [])[:50]})
|
||
# 重启后内存为空：回退数据库最近一批推荐
|
||
import sqlite3
|
||
db = str(ROOT / 'data' / 'a_stock.db')
|
||
try:
|
||
conn = sqlite3.connect(db, check_same_thread=False)
|
||
latest_ts = conn.execute(
|
||
"SELECT ts FROM quant_recommendation ORDER BY id DESC LIMIT 1").fetchone()
|
||
rows = []
|
||
if latest_ts:
|
||
rows = conn.execute(
|
||
"SELECT code,name,price,score,stars,buy_low,buy_high,"
|
||
"expected_return_pct,reason FROM quant_recommendation "
|
||
"WHERE ts=? ORDER BY score DESC LIMIT 50", (latest_ts[0],)).fetchall()
|
||
conn.close()
|
||
except Exception:
|
||
rows = []
|
||
items = [{'code': r[0], 'name': r[1] or r[0], 'price': r[2], 'score': r[3],
|
||
'stars': r[4], 'buy_low': r[5], 'buy_high': r[6],
|
||
'expected_return_pct': r[7], 'reason': r[8]} for r in rows]
|
||
return web.json_response({'ts': latest_ts[0] if rows else None, 'list': items,
|
||
'source': 'db'})
|
||
|
||
|
||
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)
|