02-指数估值查询

指数行情、估值与历史分位查询。趋势定投的核心决策依据。

当前估值快照

获取某指数最新估值与分位：

from db import get_connection
from query_utils import query_to_df, quality_filter

conn = get_connection()
df = query_to_df(conn, f"""
    SELECT d.index_name, v.trade_date, v.pe, v.pe_percentile,
           v.pb, v.pb_percentile, v.dividend_yield
    FROM index_valuation v
    JOIN dim_index d ON v.index_code = d.index_code
    WHERE v.index_code = ?
      AND {quality_filter("v")}
    ORDER BY v.trade_date DESC LIMIT 1
""", ("000300",))
print(df)

对应 SQL：

SELECT d.index_name, v.trade_date, v.pe, v.pe_percentile,
       v.pb, v.pb_percentile, v.dividend_yield
FROM index_valuation v
JOIN dim_index d ON v.index_code = d.index_code
WHERE v.index_code = '000300'
  AND v.data_quality IN ('ok', 'manual_fixed')
ORDER BY v.trade_date DESC LIMIT 1;

分位解读：

pe_percentile	估值区域	定投建议
< 0.2	极低	加大定投
0.2 - 0.4	低	正常定投
0.4 - 0.6	中	正常定投
0.6 - 0.8	高	减半定投
> 0.8	极高	暂停定投或分批止盈

全部指数估值对比

一次性查看所有关注指数的当前估值：

SELECT d.index_code, d.index_name, d.category,
       v.trade_date, v.pe, v.pe_percentile,
       v.pb, v.pb_percentile, v.dividend_yield
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')
ORDER BY v.pe_percentile ASC;

按 PE 分位升序排列，分位低的指数排在前面，便于发现低估机会。

历史估值序列

获取某指数的历史估值序列，用于绘制估值曲线：

df = query_to_df(conn, f"""
    SELECT trade_date, pe, pe_percentile, pb, pb_percentile
    FROM index_valuation
    WHERE index_code = ?
      AND trade_date >= date('now', '-10 years')
      AND {quality_filter()}
    ORDER BY trade_date
""", ("000300",))

Python 绘图示例：

import matplotlib.pyplot as plt

fig, ax1 = plt.subplots(figsize=(12, 5))
ax1.plot(df["trade_date"], df["pe"], color="steelblue", label="PE")
ax1.set_ylabel("PE", color="steelblue")
ax2 = ax1.twinx()
ax2.plot(df["trade_date"], df["pe_percentile"], color="coral", label="PE 分位")
ax2.axhline(0.2, ls="--", color="green", alpha=0.5)
ax2.axhline(0.8, ls="--", color="red", alpha=0.5)
ax2.set_ylabel("分位", color="coral")
plt.title("沪深300 估值历史")
plt.show()

估值分位历史分布

查看某指数 PE 分位的分布，判断当前处于什么位置：

SELECT
    CASE
        WHEN pe_percentile < 0.1 THEN '0-10%'
        WHEN pe_percentile < 0.2 THEN '10-20%'
        WHEN pe_percentile < 0.3 THEN '20-30%'
        WHEN pe_percentile < 0.4 THEN '30-40%'
        WHEN pe_percentile < 0.5 THEN '40-50%'
        WHEN pe_percentile < 0.6 THEN '50-60%'
        WHEN pe_percentile < 0.7 THEN '60-70%'
        WHEN pe_percentile < 0.8 THEN '70-80%'
        WHEN pe_percentile < 0.9 THEN '80-90%'
        ELSE '90-100%'
    END AS bucket,
    COUNT(*) AS days
FROM index_valuation
WHERE index_code = '000300'
  AND pe_percentile IS NOT NULL
  AND data_quality IN ('ok', 'manual_fixed')
GROUP BY bucket
ORDER BY bucket;

指数日线行情

获取日线数据，用于技术分析或趋势确认：

df = query_to_df(conn, f"""
    SELECT trade_date, open, close, high, low, volume, pct_change
    FROM index_daily
    WHERE index_code = ?
      AND trade_date >= date('now', '-1 year')
      AND {quality_filter()}
    ORDER BY trade_date
""", ("000300",))

均线计算

df["ma20"] = df["close"].rolling(20).mean()
df["ma60"] = df["close"].rolling(60).mean()
df["ma250"] = df["close"].rolling(250).mean()

趋势判断

# 多头排列：MA20 > MA60 > MA250
df["bull_alignment"] = (df["ma20"] > df["ma60"]) & (df["ma60"] > df["ma250"])

多指数相关性

分析指数间的相关性，用于组合配置：

-- 宽基指数收益率相关性
WITH returns AS (
    SELECT
        a.trade_date,
        a.pct_change AS hs300,
        b.pct_change AS zz500,
        c.pct_change AS cybz
    FROM index_daily a
    JOIN index_daily b ON a.trade_date = b.trade_date AND b.index_code = '000905'
    JOIN index_daily c ON a.trade_date = c.trade_date AND c.index_code = '399006'
    WHERE a.index_code = '000300'
      AND a.data_quality IN ('ok', 'manual_fixed')
      AND a.trade_date >= date('now', '-3 years')
)
SELECT
    AVG(hs300 * zz500) - AVG(hs300) * AVG(zz500) AS hs300_zz500_cov,
    AVG(hs300 * cybz) - AVG(hs300) * AVG(cybz) AS hs300_cybz_cov
FROM returns;

更完整的相关性矩阵建议在 Python 中用 df.corr() 计算。

封装函数建议

在 query_utils.py 中可封装以下函数：

def get_index_valuation_now(conn, index_code: str) -> pd.DataFrame:
    """获取指数最新估值。"""

def get_index_valuation_history(conn, index_code: str, years: int = 10) -> pd.DataFrame:
    """获取指数历史估值序列。"""

def get_all_index_percentiles(conn) -> pd.DataFrame:
    """获取所有指数当前估值分位，按分位升序。"""

def get_index_daily(conn, index_code: str, start_date: str = None,
                    end_date: str = None) -> pd.DataFrame:
    """获取指数日线行情。"""

具体实现见 query_utils.py（可按需创建）。

#02-指数估值查询

#当前估值快照

#全部指数估值对比

#历史估值序列

#估值分位历史分布

#指数日线行情

#均线计算

#趋势判断

#多指数相关性

#封装函数建议