趋势定投之路

01-连接与工具函数

查询接口的基础设施：数据库连接、通用工具函数。所有业务查询函数依赖此模块。

数据库连接

复用 db.py 的 get_connection，自动配置 WAL 模式与外键约束：

import sys
from pathlib import Path

# 把 02-数据脚本 加入路径
sys.path.insert(0, str(Path(__file__).parent.parent / "02-数据脚本"))

from db import get_connection

conn = get_connection()
# 执行查询
df = conn.execute("SELECT * FROM dim_index LIMIT 5").fetchall()
conn.close()

推荐用上下文管理器确保连接关闭：

from contextlib import contextmanager

@contextmanager
def get_db():
    conn = get_connection()
    try:
        yield conn
    finally:
        conn.close()

with get_db() as conn:
    df = conn.execute("SELECT COUNT(*) FROM index_daily").fetchone()

通用工具函数

建议在 query_utils.py 中封装以下通用工具（可按需创建）：

结果转 DataFrame

import pandas as pd

def query_to_df(conn, sql: str, params: tuple = None) -> pd.DataFrame:
    """执行 SQL 并返回 DataFrame。"""
    return pd.read_sql_query(sql, conn, params=params or ())

日期范围工具

from datetime import datetime, timedelta

def date_n_years_ago(n: int) -> str:
    """返回 n 年前的日期字符串 YYYY-MM-DD。"""
    return (datetime.now() - timedelta(days=365 * n)).strftime("%Y-%m-%d")

def last_n_trade_dates(conn, n: int) -> list:
    """返回最近 n 个交易日（基于 index_daily 推断）。"""
    rows = conn.execute(
        "SELECT DISTINCT trade_date FROM index_daily "
        "ORDER BY trade_date DESC LIMIT ?", (n,)
    ).fetchall()
    return [r["trade_date"] for r in rows]

数据质量过滤

def quality_filter(table_alias: str = "") -> str:
    """返回数据质量过滤条件，默认只看正常数据。"""
    prefix = f"{table_alias}." if table_alias else ""
    return f"{prefix}data_quality IN ('ok', 'manual_fixed')"

使用示例：

sql = f"""
    SELECT trade_date, close FROM index_daily
    WHERE index_code = ?
      AND {quality_filter()}
    ORDER BY trade_date DESC LIMIT 30
"""
df = query_to_df(conn, sql, ("000300",))

常用查询模板

查看标的列表

-- 所有指数
SELECT index_code, index_name, category FROM dim_index ORDER BY category, index_code;

-- 所有基金
SELECT fund_code, fund_name, fund_type FROM dim_fund ORDER BY fund_type, fund_code;

-- 所有股票
SELECT stock_code, stock_name, exchange FROM dim_stock ORDER BY exchange, stock_code;

-- 所有行业
SELECT industry_code, industry_name, industry_level FROM dim_industry ORDER BY industry_level, industry_code;

查看数据覆盖范围

SELECT
    'index_daily' AS tbl,
    COUNT(*) AS rows,
    MIN(trade_date) AS first_date,
    MAX(trade_date) AS last_date
FROM index_daily
UNION ALL
SELECT 'fund_daily', COUNT(*), MIN(trade_date), MAX(trade_date) FROM fund_daily
UNION ALL
SELECT 'stock_daily', COUNT(*), MIN(trade_date), MAX(trade_date) FROM stock_daily
UNION ALL
SELECT 'index_valuation', COUNT(*), MIN(trade_date), MAX(trade_date) FROM index_valuation;

查看更新失败记录

-- 最近一次更新的失败情况
SELECT fetcher, target_code, api_name, error_type, error_message, occurred_at
FROM update_failure_log
WHERE run_id = (SELECT MAX(run_id) FROM update_failure_log)
ORDER BY occurred_at DESC;

查看交叉验证不一致

-- 最近不一致的交叉验证记录
SELECT target_code, trade_date, field_name,
       value_primary, value_secondary, diff_pct, status, checked_at
FROM cross_validation_log
WHERE status = 'mismatch'
ORDER BY checked_at DESC
LIMIT 50;

站点集成建议

Rspress 是 Node.js 环境，无法直接调用 Python 查询。推荐两种集成方式：

方式一：预计算 JSON（推荐）

Python 脚本预计算常用指标，输出 JSON 到 public/data/ 目录，Rspress 页面通过 fetch 读取：

# 在 query_utils.py 中增加导出函数
import json

def export_index_percentiles(conn, output_path: str):
    """导出各指数当前估值分位到 JSON。"""
    df = pd.read_sql_query("""
        SELECT v.index_code, d.index_name, v.trade_date,
               v.pe, v.pe_percentile, v.pb, v.pb_percentile
        FROM index_valuation v
        JOIN dim_index d ON v.index_code = d.index_code
        WHERE v.trade_date = (SELECT MAX(trade_date) FROM index_valuation)
          AND v.data_quality IN ('ok', 'manual_fixed')
    """, conn)
    df.to_json(output_path, orient="records", force_ascii=False, indent=2)

Rspress 页面读取：

import { useEffect, useState } from "react";

export default function IndexPercentile() {
  const [data, setData] = useState([]);
  useEffect(() => {
    fetch("/data/index_percentiles.json")
      .then(res => res.json())
      .then(setData);
  }, []);
  // 渲染表格...
}

方式二：内嵌静态表格

分析文档中直接内嵌查询结果（Markdown 表格），适合不频繁更新的数据。在文档中注明数据快照时间。

趋势定投分析以中低频决策为主，方式一足够覆盖，无需引入 API 服务。