06-定投决策查询

围绕趋势定投决策场景的综合性查询：估值分位定投信号、组合分析、历史回测。整合各表数据，直接输出决策依据。

定投信号查询

单指数定投信号

根据估值分位给出定投强度建议：

SELECT
    d.index_code,
    d.index_name,
    v.trade_date,
    v.pe,
    v.pe_percentile,
    v.pb_percentile,
    CASE
        WHEN v.pe_percentile < 0.2 THEN '加大定投'
        WHEN v.pe_percentile < 0.4 THEN '正常定投'
        WHEN v.pe_percentile < 0.6 THEN '正常定投'
        WHEN v.pe_percentile < 0.8 THEN '减半定投'
        WHEN v.pe_percentile >= 0.8 THEN '暂停或止盈'
        ELSE '数据不足'
    END AS signal,
    CASE
        WHEN v.pe_percentile < 0.2 THEN 1.5
        WHEN v.pe_percentile < 0.4 THEN 1.0
        WHEN v.pe_percentile < 0.6 THEN 1.0
        WHEN v.pe_percentile < 0.8 THEN 0.5
        WHEN v.pe_percentile >= 0.8 THEN 0.0
        ELSE NULL
    END AS multiplier
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;

multiplier 表示定投倍数：1.5 表示投 1.5 倍基础金额，0.5 表示减半，0 表示暂停。

组合定投信号

from db import get_connection
from query_utils import query_to_df

conn = get_connection()

# 关注组合
portfolio = ["000300", "000905", "399006", "000688"]

df = query_to_df(conn, """
    SELECT
        d.index_code, d.index_name,
        v.trade_date, v.pe, v.pe_percentile,
        CASE
            WHEN v.pe_percentile < 0.2 THEN 1.5
            WHEN v.pe_percentile < 0.6 THEN 1.0
            WHEN v.pe_percentile < 0.8 THEN 0.5
            ELSE 0.0
        END AS multiplier
    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')
      AND d.index_code IN (%s)
    ORDER BY v.pe_percentile
""" % ",".join(f"'{c}'" for c in portfolio))

print(df)
# 计算本次定投总额（假设基础金额 1000 元/只）
base = 1000
total = (df["multiplier"] * base).sum()
print(f"本次定投总额: {total:.0f} 元")

微笑曲线识别

查找当前处于"微笑曲线"底部区域的指数（持续下跌后估值低位）：

WITH price_stats AS (
    SELECT
        t.index_code,
        d.index_name,
        MAX(CASE WHEN t.trade_date = (SELECT MAX(trade_date) FROM index_daily) THEN t.close END) AS latest_close,
        MAX(t.close) AS all_time_high,
        MAX(CASE WHEN t.trade_date >= date('now', '-250 days') THEN t.close END) AS recent_high
    FROM index_daily t
    JOIN dim_index d ON t.index_code = d.index_code
    WHERE t.data_quality IN ('ok', 'manual_fixed')
    GROUP BY t.index_code, d.industry_name
)
SELECT
    p.index_name,
    p.latest_close,
    p.all_time_high,
    (p.latest_close / p.all_time_high - 1) * 100 AS drawdown_from_high,
    v.pe_percentile
FROM price_stats p
JOIN index_valuation v
    ON p.index_code = v.index_code
    AND v.trade_date = (SELECT MAX(trade_date) FROM index_valuation)
WHERE v.pe_percentile < 0.3
  AND v.data_quality IN ('ok', 'manual_fixed')
ORDER BY drawdown_from_high ASC;

满足"从高点跌幅大 + 当前估值分位低"的标的，适合定投积累筹码。

组合配置分析

资产配置建议（基于估值）

SELECT
    d.category,
    d.index_name,
    v.pe_percentile,
    CASE
        WHEN v.pe_percentile < 0.3 THEN '超配'
        WHEN v.pe_percentile < 0.7 THEN '标配'
        ELSE '低配'
    END AS allocation
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 d.category, v.pe_percentile;

定投回测

简单定投回测

def backtest_dca(conn, index_code: str, start_date: str,
                 monthly_amount: float = 1000) -> dict:
    """定投回测：每月固定金额投入。

    返回：累计投入、当前市值、收益率、年化收益率
    """
    df = query_to_df(conn, """
        SELECT trade_date, close FROM index_daily
        WHERE index_code = ? AND trade_date >= ?
          AND data_quality IN ('ok', 'manual_fixed')
        ORDER BY trade_date
    """, (index_code, start_date))

    # 每月第一个交易日扣款
    df["ym"] = df["trade_date"].str[:7]
    monthly = df.groupby("ym").first().reset_index()

    total_invested = 0.0
    total_shares = 0.0
    for _, row in monthly.iterrows():
        shares = monthly_amount / row["close"]
        total_shares += shares
        total_invested += monthly_amount

    latest_close = df.iloc[-1]["close"]
    current_value = total_shares * latest_close
    total_return = (current_value / total_invested - 1) * 100

    # 年化
    days = (df.iloc[-1]["trade_date"] - df.iloc[0]["trade_date"]).days
    years = max(days / 365, 0.01)
    annual_return = ((current_value / total_invested) ** (1 / years) - 1) * 100

    return {
        "total_invested": total_invested,
        "current_value": current_value,
        "total_return_pct": total_return,
        "annual_return_pct": annual_return,
        "months": len(monthly),
    }

估值分位定投回测

def backtest_valuation_dca(conn, index_code: str, start_date: str,
                           base_amount: float = 1000) -> dict:
    """估值分位定投：低估值多投，高估值少投。"""
    df = query_to_df(conn, """
        SELECT t.trade_date, t.close, v.pe_percentile
        FROM index_daily t
        LEFT JOIN index_valuation v
            ON t.index_code = v.index_code AND t.trade_date = v.trade_date
        WHERE t.index_code = ? AND t.trade_date >= ?
          AND t.data_quality IN ('ok', 'manual_fixed')
        ORDER BY t.trade_date
    """, (index_code, start_date))

    df["ym"] = df["trade_date"].str[:7]
    monthly = df.groupby("ym").first().reset_index()

    total_invested = 0.0
    total_shares = 0.0
    for _, row in monthly.iterrows():
        pct = row["pe_percentile"]
        if pct is None or pct < 0.2:
            amount = base_amount * 1.5
        elif pct < 0.6:
            amount = base_amount * 1.0
        elif pct < 0.8:
            amount = base_amount * 0.5
        else:
            continue  # 暂停

        shares = amount / row["close"]
        total_shares += shares
        total_invested += amount

    latest_close = df.iloc[-1]["close"]
    current_value = total_shares * latest_close
    return {
        "total_invested": total_invested,
        "current_value": current_value,
        "total_return_pct": (current_value / total_invested - 1) * 100,
    }

对比两种策略的收益差异，验证估值分位定投是否跑赢简单定投。详细回测方法论见 04-终点段/02-定投策略研究。

止盈信号查询

估值止盈

SELECT
    d.index_code, d.index_name,
    v.trade_date, v.pe, v.pe_percentile,
    CASE
        WHEN v.pe_percentile >= 0.9 THEN '强烈止盈'
        WHEN v.pe_percentile >= 0.8 THEN '分批止盈'
        WHEN v.pe_percentile >= 0.7 THEN '关注止盈'
        ELSE '持有'
    END AS stop_profit_signal
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 DESC;

止盈策略的完整说明见 04-终点段/02-定投策略研究/12-定投止盈止损具体策略。

决策日志支持

生成定投决策记录

def generate_decision_log(conn, portfolio: list, base_amount: float = 1000) -> list:
    """生成当日定投决策日志条目。"""
    df = query_to_df(conn, """
        SELECT
            d.index_code, d.index_name,
            v.pe_percentile,
            CASE
                WHEN v.pe_percentile < 0.2 THEN 1.5
                WHEN v.pe_percentile < 0.6 THEN 1.0
                WHEN v.pe_percentile < 0.8 THEN 0.5
                ELSE 0.0
            END AS multiplier
        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')
          AND d.index_code IN (%s)
    """ % ",".join(f"'{c}'" for c in portfolio))

    records = []
    for _, row in df.iterrows():
        records.append({
            "index_code": row["index_code"],
            "index_name": row["index_name"],
            "pe_percentile": row["pe_percentile"],
            "amount": base_amount * row["multiplier"],
            "action": "定投" if row["multiplier"] > 0 else "暂停",
        })
    return records

生成的决策记录可直接写入 04-终点段/05-决策复盘/02-买入决策日志。

封装函数建议

def get_dca_signals(conn, index_codes: list) -> pd.DataFrame:
    """获取组合定投信号。"""

def backtest_dca_strategy(conn, index_code: str, start_date: str,
                          strategy: str = "simple") -> dict:
    """定投策略回测（simple/valuation）。"""

def get_stop_profit_signals(conn, index_codes: list) -> pd.DataFrame:
    """获取止盈信号。"""

#06-定投决策查询

#定投信号查询

#单指数定投信号

#组合定投信号

#微笑曲线识别

#组合配置分析

#资产配置建议（基于估值）

#定投回测

#简单定投回测

#估值分位定投回测

#止盈信号查询

#估值止盈

#决策日志支持

#生成定投决策记录

#封装函数建议

06-定投决策查询

定投信号查询

单指数定投信号

组合定投信号

微笑曲线识别

组合配置分析

资产配置建议（基于估值）

定投回测

简单定投回测

估值分位定投回测

止盈信号查询

估值止盈

决策日志支持

生成定投决策记录

封装函数建议