simon
fce725e13c
从 Point-in-Time 股票筛选到统一 Web 前端的完整链路： 筛选 → 画像 → 策略 → 回测 → Walk-forward → 绩效分析 → 报告/前端。 架构 - 数据层与策略层分离；策略代码不写 SQL，只经 data/repo.py 取数 - 所有业务阈值集中在 config/*.yml，代码零硬编码（字段写错直接报错） - 报告只做「run_id → SQL → 渲染」，不做任何计算，数字可追溯 - 前后端分离：output/ 静态站点 + hdiv web 提供的 REST API 数据安全 - 只增不删：SQL 钩子拦截 DELETE/DROP/TRUNCATE，并有源码扫描测试守护 - qlib 原有表只读，本项目数据写入 hd_ 前缀表 - 回补使用 INSERT IGNORE，保证既有行零改动 - .env 存密钥且已 gitignore；output/、logs/、.venv/ 不入库 交付物 - 30 张 hd_* 表、7 个 YAML 配置、283 项自动化测试 - 统一 Web 前端（hash 路由 SPA）+ nginx 部署配置与 launchd 托管脚本 如实声明的限制 - 策略缺少稳定的样本外超额收益（Walk-forward 7 窗口均值 -0.95%， 基准 +2.29%）；其价值体现在回撤控制，而非超额收益 - 涨跌停/停牌约束仅覆盖 2019 年起；index_weight 尚未填充 - AI Agent 层（plan.md 第四版 P8）未实现 详见 docs/user-guide.md 与 docs/implementation-status.md。
204 lines
7.0 KiB
Python
204 lines
7.0 KiB
Python
"""表结构与幂等性测试（对应 development-plan.md §7.2 硬约束）。
|
|
|
|
需要真实数据库的用例标记为 ``db``；纯结构校验的用例无需数据库。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
import pytest
|
|
|
|
from hdiv.core.config import load_config
|
|
from hdiv.core.errors import SafetyViolation
|
|
from hdiv.data.schema import ALL_TABLES, TABLE_NAMES
|
|
|
|
CFG = load_config("datasource")
|
|
|
|
# 需要库的表清单（无库时跳过）
|
|
_PLAIN_TESTS_NEED_DB = pytest.mark.db
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 纯结构校验（无需数据库）
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_table_count() -> None:
|
|
assert len(ALL_TABLES) == 30
|
|
assert len(set(TABLE_NAMES)) == 30
|
|
|
|
|
|
def test_all_tables_use_own_prefix() -> None:
|
|
bad = [t.name for t in ALL_TABLES if not t.name.startswith(CFG.database.own_prefix)]
|
|
assert not bad, f"以下表未使用自有前缀：{bad}"
|
|
|
|
|
|
def test_ddl_never_contains_destructive_statements() -> None:
|
|
destructive = re.compile(r"\b(DROP|TRUNCATE|DELETE)\b", re.IGNORECASE)
|
|
for t in ALL_TABLES:
|
|
assert not destructive.search(t.ddl), f"{t.name} 的 DDL 含破坏性语句"
|
|
for m in t.migrations:
|
|
assert not re.search(r"\b(DROP\s+TABLE|TRUNCATE|DELETE)\b", m.sql, re.IGNORECASE), (
|
|
f"{t.name} 迁移 {m.name} 含破坏性语句：{m.sql}"
|
|
)
|
|
|
|
|
|
def test_create_statements_are_idempotent() -> None:
|
|
for t in ALL_TABLES:
|
|
assert "CREATE TABLE IF NOT EXISTS" in t.ddl, f"{t.name} 未使用 IF NOT EXISTS"
|
|
|
|
|
|
def test_every_table_has_primary_key() -> None:
|
|
for t in ALL_TABLES:
|
|
assert "PRIMARY KEY" in t.ddl, f"{t.name} 缺少主键"
|
|
|
|
|
|
def test_tables_have_comment() -> None:
|
|
for t in ALL_TABLES:
|
|
assert t.comment, f"{t.name} 缺少注释"
|
|
assert "COMMENT='" in t.ddl or 'COMMENT="' in t.ddl, f"{t.name} 的 DDL 缺少表注释"
|
|
|
|
|
|
def test_no_nullable_column_in_unique_key() -> None:
|
|
"""MySQL 唯一约束不约束 NULL —— 参与唯一键的列必须 NOT NULL。
|
|
|
|
唯一例外是显式构造的 dedup_key（本身 NOT NULL）。
|
|
"""
|
|
problems: list[str] = []
|
|
for t in ALL_TABLES:
|
|
for m in re.finditer(r"UNIQUE KEY `\w+` \(([^)]+)\)", t.ddl):
|
|
cols = [c.strip().strip("`") for c in m.group(1).split(",")]
|
|
for col in cols:
|
|
# 找到该列的定义行
|
|
pat = re.compile(rf"^\s*`{re.escape(col)}`\s+([A-Za-z]+(?:\([^)]*\))?)(.*)$", re.M)
|
|
mm = pat.search(t.ddl)
|
|
if not mm:
|
|
problems.append(f"{t.name}.{col} 未找到列定义")
|
|
continue
|
|
rest = mm.group(2)
|
|
if "NOT NULL" not in rest:
|
|
problems.append(
|
|
f"{t.name}.{col} 可空却参与唯一键 —— NULL 会绕过唯一约束造成重复"
|
|
)
|
|
assert not problems, "\n".join(problems)
|
|
|
|
|
|
def test_character_set_is_utf8mb4() -> None:
|
|
for t in ALL_TABLES:
|
|
assert "utf8mb4" in t.ddl, f"{t.name} 未使用 utf8mb4"
|
|
|
|
|
|
def test_engine_is_innodb() -> None:
|
|
for t in ALL_TABLES:
|
|
assert "InnoDB" in t.ddl, f"{t.name} 未使用 InnoDB"
|
|
|
|
|
|
def test_migration_names_are_unique_within_table() -> None:
|
|
for t in ALL_TABLES:
|
|
names = [m.name for m in t.migrations]
|
|
assert len(names) == len(set(names)), f"{t.name} 迁移名重复：{names}"
|
|
|
|
|
|
def test_migrations_do_not_add_dropped_tables() -> None:
|
|
"""迁移只允许 ADD/MODIFY，不允许改表名或删列。"""
|
|
for t in ALL_TABLES:
|
|
for m in t.migrations:
|
|
assert not re.search(r"\bDROP\s+COLUMN\b", m.sql, re.IGNORECASE), m.sql
|
|
assert not re.search(r"\bRENAME\b", m.sql, re.IGNORECASE), m.sql
|
|
|
|
|
|
def test_own_prefix_guard_rejects_foreign_table() -> None:
|
|
from hdiv.data.ddl import _assert_own_prefix
|
|
|
|
with pytest.raises(SafetyViolation):
|
|
_assert_own_prefix(CFG, "stock")
|
|
with pytest.raises(SafetyViolation):
|
|
_assert_own_prefix(CFG, "hd_ok") if False else _assert_own_prefix(CFG, "alembic_version")
|
|
|
|
|
|
def test_own_prefix_guard_accepts_own_table() -> None:
|
|
from hdiv.data.ddl import _assert_own_prefix
|
|
|
|
_assert_own_prefix(CFG, "hd_dividend") # 不应抛错
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 需要数据库
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@_PLAIN_TESTS_NEED_DB
|
|
def test_ddl_plan_is_idempotent_on_live_db() -> None:
|
|
"""连续两次 plan 都应返回 0 个动作（结构已就绪且迁移已应用）。"""
|
|
pytest.importorskip("pymysql")
|
|
from hdiv.data import ddl, db
|
|
|
|
db.load_dotenv_once()
|
|
try:
|
|
db.list_tables(CFG)
|
|
except Exception as exc: # pragma: no cover - 环境相关
|
|
pytest.skip(f"数据库不可用：{exc}")
|
|
|
|
actions = ddl.plan(CFG)
|
|
assert actions == [], f"结构未收敛，仍待执行：{[(a.kind, a.table, a.detail) for a in actions]}"
|
|
|
|
|
|
@_PLAIN_TESTS_NEED_DB
|
|
def test_all_declared_tables_exist() -> None:
|
|
pytest.importorskip("pymysql")
|
|
from hdiv.data import db
|
|
|
|
db.load_dotenv_once()
|
|
try:
|
|
existing = set(db.list_tables(CFG))
|
|
except Exception as exc: # pragma: no cover
|
|
pytest.skip(f"数据库不可用：{exc}")
|
|
missing = [t for t in TABLE_NAMES if t not in existing]
|
|
assert not missing, f"库中缺表：{missing}"
|
|
|
|
|
|
@_PLAIN_TESTS_NEED_DB
|
|
def test_verify_reports_no_problems() -> None:
|
|
pytest.importorskip("pymysql")
|
|
from hdiv.data import db, ddl
|
|
|
|
db.load_dotenv_once()
|
|
try:
|
|
problems = ddl.verify(CFG)
|
|
except Exception as exc: # pragma: no cover
|
|
pytest.skip(f"数据库不可用：{exc}")
|
|
assert not problems, "\n".join(problems)
|
|
|
|
|
|
@_PLAIN_TESTS_NEED_DB
|
|
def test_alembic_version_untouched() -> None:
|
|
"""决策 D1：本项目不得修改 qlib 的 Alembic 版本链。"""
|
|
pytest.importorskip("pymysql")
|
|
from hdiv.data import db
|
|
|
|
db.load_dotenv_once()
|
|
try:
|
|
df = db.read_sql("SELECT version_num FROM alembic_version", cfg=CFG)
|
|
except Exception as exc: # pragma: no cover
|
|
pytest.skip(f"数据库不可用：{exc}")
|
|
# 只要存在且可读即可；本项目从未写入该表（由 SafetyViolation 保证）
|
|
assert len(df) >= 1
|
|
|
|
|
|
@_PLAIN_TESTS_NEED_DB
|
|
def test_project_created_tables_all_have_own_prefix() -> None:
|
|
"""本项目建的表必须 100% 匹配 hd_ 前缀。"""
|
|
pytest.importorskip("pymysql")
|
|
from hdiv.data import db
|
|
from hdiv.data.schema import TABLE_NAMES as declared
|
|
|
|
db.load_dotenv_once()
|
|
try:
|
|
existing = set(db.list_tables(CFG))
|
|
except Exception as exc: # pragma: no cover
|
|
pytest.skip(f"数据库不可用：{exc}")
|
|
hd_tables = {t for t in existing if t.startswith(CFG.database.own_prefix)}
|
|
assert hd_tables <= set(declared), f"库中存在未声明的 hd_ 表：{hd_tables - set(declared)}"
|
|
assert set(declared) <= existing, f"声明的表未全部创建：{set(declared) - existing}"
|