30 lines
1.4 KiB
Python
30 lines
1.4 KiB
Python
# -*- 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)
|