cirry
528357c3f5
- 首页双入口（智能选股/策略回测）：引入 vue-router，顶部导航 - 智能选股：自然语言 -> LLM 解析结构化条件（智谱 GLM，OpenAI 兼容，/v4 兼容）-> SQL 快照预筛 + pandas 指标过滤（复用 indicators 单一事实源） - 条件模型：指标 vs 常数/指标（value_indicator，如 DIF>DEA、close<布林下轨）、lookback+match 表达连续N天/近N天任一天、市值/PE/PB/换手率快照条件、默认排除 ST/退市/北交所 - 全市场数据同步：按 trade_date 批量拉取未复权日线（与回测 candles qfq 隔离），交易日历/股票列表本地缓存，daily_basic 仅最新截面，Tushare 限频兜底（分钟级重试/小时级降级） - 存储：DATABASE_URL 切远程 PostgreSQL（cirry.cn/stock），本地 SQLite 已移除 - .env 入库（私有仓库）；smoke_test 扩展选股链路 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
101 lines
3.9 KiB
Python
101 lines
3.9 KiB
Python
"""开发自检脚本：跑一遍 /health 与 /backtest，打印结果。
|
||
|
||
用 FastAPI TestClient（无需起服务器，同进程验证全链路）。
|
||
用法： uv run --with httpx --directory backend python smoke_test.py
|
||
"""
|
||
import json
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.main import app
|
||
|
||
# 必须用 with：lifespan（建表）只在进入上下文时执行
|
||
with TestClient(app) as c:
|
||
r = c.get("/api/health")
|
||
print("== /api/health ==", r.status_code, r.json())
|
||
|
||
r = c.post(
|
||
"/api/backtest",
|
||
json={
|
||
"symbol": "DEMO",
|
||
"strategy": "macd_cross",
|
||
"params": {"fast": 12, "slow": 26, "signal": 9},
|
||
"initial_cash": 100000.0,
|
||
"fast_mode": False,
|
||
},
|
||
)
|
||
print("== /api/backtest ==", r.status_code)
|
||
if r.status_code != 200:
|
||
print("ERROR:", r.text)
|
||
raise SystemExit(1)
|
||
|
||
d = r.json()
|
||
print("candles :", len(d["candles"]))
|
||
print("signals :", len(d["signals"]), "(买卖点)")
|
||
print("equity pts :", len(d["equity"]))
|
||
print("final_cash :", round(d["final_cash"], 2))
|
||
print("final_pos :", d["final_position"])
|
||
print("metrics :", json.dumps(d["metrics"], ensure_ascii=False, indent=2))
|
||
print("first signal :", d["signals"][0] if d["signals"] else None)
|
||
|
||
assert len(d["candles"]) > 100
|
||
|
||
# 周期聚合：周线 K 线数应明显少于日线
|
||
rw = c.post(
|
||
"/api/backtest",
|
||
json={"symbol": "DEMO", "timeframe": "1w", "strategy": "macd_cross",
|
||
"params": {"fast": 12, "slow": 26, "signal": 9}, "initial_cash": 100000.0},
|
||
)
|
||
wd = rw.json()
|
||
print("== weekly ==", rw.status_code, "candles:", len(wd["candles"]), "vs daily", len(d["candles"]))
|
||
assert rw.status_code == 200
|
||
assert len(wd["candles"]) < len(d["candles"])
|
||
|
||
print("\n✅ 后端全链路自检通过（含周期聚合）")
|
||
|
||
# ---------- 智能选股 ----------
|
||
print("\n== 智能选股 ==")
|
||
|
||
# 1) JSON 提取容错（不联网）：代码围栏 / 多余文本
|
||
from app.screener.llm import _extract_json
|
||
assert _extract_json('```json\n{"a": 1}\n```') == {"a": 1}
|
||
assert _extract_json('好的，结果如下：{"indicator": [], "snapshot": []} 谢谢')["indicator"] == []
|
||
print("_extract_json 围栏/噪音容错 ✅")
|
||
|
||
# 2) 未配置 LLM_API_KEY 时 /run 返回 503（确定性，不联网）
|
||
from app.config import settings as _s
|
||
r = c.post("/api/screener/run", json={"text": "这两天 KDJ 的 J 小于 10"})
|
||
if not _s.llm_api_key:
|
||
assert r.status_code == 503, f"无 key 应 503，实际 {r.status_code}"
|
||
print("无 LLM_API_KEY -> 503 ✅")
|
||
else:
|
||
print("已配置 LLM_API_KEY，跳过 503 用例")
|
||
|
||
# 3) 直传条件选股（不依赖 LLM；依赖已同步的全市场数据）
|
||
from app.models import MarketDaily # noqa: F401
|
||
from sqlalchemy import select, func
|
||
from app.db import async_session
|
||
import asyncio
|
||
|
||
async def _has_data() -> bool:
|
||
async with async_session() as session:
|
||
return (await session.scalar(select(func.count()).select_from(MarketDaily))) or 0 > 0
|
||
|
||
if asyncio.run(_has_data()):
|
||
r = c.post("/api/screener/run", json={
|
||
"text": "测试直传",
|
||
"conditions": {
|
||
"indicator": [{"indicator": "kdj_j", "params": {"n": 9, "m1": 3, "m2": 3},
|
||
"op": "lt", "value": 0, "lookback": 1, "match": "all"}],
|
||
"snapshot": [],
|
||
},
|
||
})
|
||
assert r.status_code == 200, f"直传选股失败 {r.status_code}: {r.text}"
|
||
d = r.json()
|
||
assert d["total"] >= 0
|
||
print(f"KDJ J<0 选股 ✅ 命中 {d['total']} 只，基准日 {(d['trade_date'] or '')[:10]}")
|
||
else:
|
||
print("（未同步全市场数据，跳过直传选股用例；运行 POST /api/screener/sync 后再试）")
|
||
|
||
print("\n✅ 智能选股自检通过")
|