claude_dev
557479b848
后端3小端点(sanguo_factor/universe_pools.py+routes):/factor/universe/pools
池清单(白名单6指数+当前成分数,in_current=1池语义≠回测并集口径)/pool/{key}
成分(code+name读constituent_unified)/search?q=(代码前缀OR名称子句DISTINCT
LIMIT10);cfg=None兜底同analyzer;+6单测+2端点测(⚠️首版误覆盖batch_eval的
universe.py,git恢复后改名universe_pools,292测全绿)
前端:①New.vue终端风重写——FactorPicker(搜索+三色分组点选+已选托盘,
240因子弃下拉)+UniversePicker(三层输入:预设池一键选[跨行业30前端常量+
6指数池走后端]/搜索combobox防抖250ms带名称候选/chips池可删+复制+清空,
批量粘贴折叠兜底)+原生date input禁未来+跨度显示+校验前置(禁用+原因文案);
要素零变化(query.factor预填/hydrate回填/≥1因子+≥2标的/混合分隔)。
②Result.vue外壳——5格metric-strip(最优ICIR正红负绿/最优因子青)+IC明细
终端表格(右对齐tabular-nums,|t|≥2加粗,最优因子行青标+左青条)+tears区
tabs终端化(TearsPanel本体不动)。③factorSamples.ts常量:跨行业30只
(code+name)详情页一键直达与New页预设共用同源(LeaderboardDetail改引,
删本地重复清单)。npm run build绿(vue-tsc+rolldown)
83 lines
2.7 KiB
Python
83 lines
2.7 KiB
Python
"""标的池查询(New 页三层输入的后端): 预设指数池 + 标的搜索.
|
|
|
|
数据源 constituent_unified(与组合回测同库同表)。池语义 = 当前成分
|
|
(in_current=1)——"沪深300 池"给用户的就是当下的 300 只,历史并集(含被踢)
|
|
是回测口径,不是选池口径。跨行业 30 只样本是前端常量(与详情页一键直达
|
|
同源),不走本模块。
|
|
"""
|
|
import sqlite3
|
|
|
|
# 池白名单:顺序即前端展示顺序(index_code, 中文名)
|
|
POOLS: list[tuple[str, str]] = [
|
|
("000300", "沪深300"),
|
|
("000905", "中证500"),
|
|
("000852", "中证1000"),
|
|
("932000", "中证2000"),
|
|
("000016", "上证50"),
|
|
("000985", "中证全A"),
|
|
]
|
|
|
|
_POOL_NAMES = dict(POOLS)
|
|
|
|
|
|
def _conn() -> sqlite3.Connection:
|
|
"""连行情库(constituent_unified 所在)。cfg 兜底同 analyzer(None→默认配置)."""
|
|
from sanguo_data.config import load_config, find_config_path
|
|
|
|
cfg = load_config(find_config_path())
|
|
vnpy_db = getattr(cfg, "data_paths", {}).get("vnpy_db")
|
|
if not vnpy_db:
|
|
raise RuntimeError("config 缺 data_paths.vnpy_db — 检查 data_platform.yaml 初始化")
|
|
return sqlite3.connect(vnpy_db, timeout=30)
|
|
|
|
|
|
def list_pools() -> list[dict]:
|
|
"""预设池清单(白名单顺序 + 当前成分数)."""
|
|
out: list[dict] = []
|
|
conn = _conn()
|
|
try:
|
|
for code, name in POOLS:
|
|
n = conn.execute(
|
|
"SELECT COUNT(*) FROM constituent_unified "
|
|
"WHERE index_code=? AND in_current=1",
|
|
(code,),
|
|
).fetchone()[0]
|
|
out.append({"key": code, "name": name, "count": int(n)})
|
|
finally:
|
|
conn.close()
|
|
return out
|
|
|
|
|
|
def pool_stocks(key: str) -> list[dict]:
|
|
"""池成分(当前, 代码升序): [{code, name}]。白名单外的 key 返回 [](合法缺失)."""
|
|
if key not in _POOL_NAMES:
|
|
return []
|
|
conn = _conn()
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT code, code_name FROM constituent_unified "
|
|
"WHERE index_code=? AND in_current=1 ORDER BY code",
|
|
(key,),
|
|
).fetchall()
|
|
finally:
|
|
conn.close()
|
|
return [{"code": str(c), "name": str(n or c)} for c, n in rows]
|
|
|
|
|
|
def search_stocks(q: str, limit: int = 10) -> list[dict]:
|
|
"""标的搜索: 代码前缀 OR 名称子串(同 code 多指数去重), LIMIT."""
|
|
v = q.strip()
|
|
if not v:
|
|
return []
|
|
conn = _conn()
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT DISTINCT code, code_name FROM constituent_unified "
|
|
"WHERE code LIKE ? OR code_name LIKE ? "
|
|
"ORDER BY code LIMIT ?",
|
|
(f"{v}%", f"%{v}%", limit),
|
|
).fetchall()
|
|
finally:
|
|
conn.close()
|
|
return [{"code": str(c), "name": str(n or c)} for c, n in rows]
|