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)
77 lines
2.6 KiB
Python
77 lines
2.6 KiB
Python
"""Tests for sanguo_factor.universe_pools - 标的池查询(New 页三层输入后端)."""
|
|
import sqlite3
|
|
import sys
|
|
import os
|
|
_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0"))
|
|
if _VNPY_SRC not in sys.path:
|
|
sys.path.insert(0, _VNPY_SRC)
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture()
|
|
def fake_db(tmp_path, monkeypatch):
|
|
"""tmp sqlite 造 constituent_unified 小样本,patch _conn."""
|
|
db = tmp_path / "uni.db"
|
|
conn = sqlite3.connect(db)
|
|
conn.execute(
|
|
"CREATE TABLE constituent_unified ("
|
|
"index_code TEXT, code TEXT, code_name TEXT, in_current INT, was_removed INT)"
|
|
)
|
|
rows = [
|
|
# 沪深300: 当前 2 只 + 被踢 1 只
|
|
("000300", "600519", "贵州茅台", 1, 0),
|
|
("000300", "000858", "五粮液", 1, 0),
|
|
("000300", "600001", "邯郸钢铁", 0, 1),
|
|
# 中证500: 当前 1 只(与 300 重叠一只以测 DISTINCT)
|
|
("000905", "600519", "贵州茅台", 1, 0),
|
|
("000905", "300750", "宁德时代", 1, 0),
|
|
# 无名称条目 → name 回退 code
|
|
("000905", "002594", None, 1, 0),
|
|
]
|
|
conn.executemany("INSERT INTO constituent_unified VALUES (?,?,?,?,?)", rows)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
import sanguo_factor.universe_pools as uni
|
|
|
|
monkeypatch.setattr(uni, "_conn", lambda: sqlite3.connect(str(db), timeout=5))
|
|
return uni
|
|
|
|
|
|
def test_list_pools_whitelist_order_and_current_count(fake_db):
|
|
pools = fake_db.list_pools()
|
|
keys = [p["key"] for p in pools]
|
|
assert keys[0] == "000300" and "000905" in keys
|
|
by_key = {p["key"]: p for p in pools}
|
|
assert by_key["000300"]["count"] == 2 # 被踢的邯郸钢铁不计
|
|
assert by_key["000905"]["count"] == 3
|
|
|
|
|
|
def test_pool_stocks_current_only(fake_db):
|
|
stocks = fake_db.pool_stocks("000300")
|
|
assert [s["code"] for s in stocks] == ["000858", "600519"] # ORDER BY code,只当前
|
|
assert stocks[1]["name"] == "贵州茅台"
|
|
|
|
|
|
def test_pool_stocks_name_fallback_to_code(fake_db):
|
|
stocks = {s["code"]: s for s in fake_db.pool_stocks("000905")}
|
|
assert stocks["002594"]["name"] == "002594" # NULL 名称回退代码
|
|
|
|
|
|
def test_pool_stocks_unknown_key_empty(fake_db):
|
|
assert fake_db.pool_stocks("999999") == []
|
|
|
|
|
|
def test_search_by_code_prefix_and_name_substr(fake_db):
|
|
by_code = {s["code"] for s in fake_db.search_stocks("6005")}
|
|
assert by_code == {"600519"}
|
|
by_name = {s["code"] for s in fake_db.search_stocks("茅台")}
|
|
assert by_name == {"600519"}
|
|
# 同股多指数 → DISTINCT 去重
|
|
assert len(fake_db.search_stocks("贵州")) == 1
|
|
|
|
|
|
def test_search_blank_returns_empty(fake_db):
|
|
assert fake_db.search_stocks(" ") == []
|