“通用指数定投机器人” 本身是一个定期定额（或定期不定额）投资指数的策略，现在要给它加上 估值分位加仓档位，并且是 自定义的 5 档梯度，也就是说根据当前指数 PE（或 PB）在历史中的分位数，来决定当期投入金额的倍率。
我会按照你给的顺序，用 Python 实现一个简化但结构清晰的版本，方便你后续扩展成实盘或回测系统。
一、实际应用场景描述
指数定投 是普通投资者参与股市最常见的方式之一，比如每周固定投 1000 元到沪深 300 ETF。
但“固定金额”有一个问题：没有考虑市场贵贱——贵的时候买和便宜的时候买一样多，长期收益不是最优。
估值分位加仓 的思路是：
- 当指数估值处于历史 低位（便宜） → 多买
- 当指数估值处于历史 高位（贵） → 少买或不买
这样能 摊薄成本、提升长期收益率，是智能投顾中常见的“增强型定投”策略。
二、引入痛点
传统固定金额定投有以下痛点：
1. 不管贵贱一个价沪深 300 PE 15 倍买 1000 元，PE 8 倍还是买 1000 元，没有利用“便宜多买”的机会。
2. 高位站岗风险如果在市场高位（如 2015 年牛市顶点）开始定投，需要很长时间回本。
3. 资金利用率低低位时没有加大投入，导致底部筹码收集不足。
4. 缺乏系统化规则很多人的“手动加仓”凭感觉，没有统一标准，容易情绪化操作。
三、核心逻辑讲解
3.1 核心思想
我们用 PE（市盈率）历史分位数 来衡量“贵贱”：
PE 分位数 = (当前 PE 在历史中排多少%)
例如：
- 当前 PE 分位 = 10% → 比历史上 90% 的时间都便宜 → 加大买入
- 当前 PE 分位 = 80% → 比历史上 80% 的时间都贵 → 减少买入或暂停
3.2 五档加仓梯度设计
我们定义 5 个档位，你可以自定义：
档位 PE 分位区间 买入倍率 含义
1 < 20% 2.0x 很度低估，重仓
2 20% ~ 40% 1.5x 低估，加码
3 40% ~ 60% 1.0x 正常，按基准投
4 60% ~ 80% 0.5x 高估，少投
5 ≥ 80% 0.0x 极度高估，暂停
倍率乘上“基准定投金额”，就是当期实际买入金额。
3.3 关键公式
当期买入金额 = 基准金额 × 档位倍率
如果档位倍率为 0，则当期不买入（相当于暂停定投）。
四、项目结构（简化版）
index_dca_robot/
├── README.md
├── requirements.txt
├── config.yaml
├── data/
│ └── index_pe_history.csv # 指数历史 PE 数据
├── src/
│ ├── data_loader.py # 数据加载
│ ├── valuation.py # ★ 估值分位计算
│ ├── dca_engine.py # ★ 定投引擎（含五档加仓）
│ └── backtester.py # 回测与统计
└── main.py
五、完整代码（模块化 + 清晰注释）
"requirements.txt"
pandas>=1.5
numpy>=1.21
pyyaml>=6.0
matplotlib>=3.5
"config.yaml"
# 指数定投机器人配置
# 定投参数
dca:
base_amount: 1000 # 基准定投金额（元）
frequency: "W" # 定投频率：W=周，M=月
# ★ 五档加仓梯度（可自定义）
valuation_brackets:
- { max_pct: 20, multiplier: 2.0 } # 极度低估
- { max_pct: 40, multiplier: 1.5 } # 低估
- { max_pct: 60, multiplier: 1.0 } # 正常
- { max_pct: 80, multiplier: 0.5 } # 高估
- { max_pct: 100, multiplier: 0.0 } # 极度高估
# 回测区间
backtest:
start_date: "2018-01-01"
end_date: "2024-12-31"
"src/data_loader.py"
"""
data_loader.py
加载指数历史 PE 数据
"""
import pandas as pd
def load_pe_history(filepath: str) -> pd.DataFrame:
"""
加载指数历史 PE 数据
预期格式:
date,pe
2018-01-02,12.5
2018-01-03,12.3
...
"""
df = pd.read_csv(filepath, parse_dates=['date'])
df = df.set_index('date').sort_index()
return df
"src/valuation.py"（★ 核心模块）
"""
valuation.py
★ 估值分位计算模块
"""
import pandas as pd
import numpy as np
class ValuationEngine:
"""
根据历史 PE 数据，计算当前 PE 的分位数，
并映射到对应的“买入倍率”。
"""
def __init__(self, brackets: list[dict]):
"""
参数:
brackets: list of dict，每个 dict 包含:
- max_pct: 分位上限（0~100）
- multiplier: 对应买入倍率
"""
# 按 max_pct 升序排列
self.brackets = sorted(brackets, key=lambda x: x['max_pct'])
def compute_percentile(self, pe_series: pd.Series, current_pe: float) -> float:
"""
计算当前 PE 在历史中的分位数（0~100）
逻辑：
(历史 PE <= 当前 PE 的天数) / (总天数) × 100
"""
if len(pe_series) == 0:
return 50.0 # 无数据时默认中位
count = (pe_series <= current_pe).sum()
pct = (count / len(pe_series)) * 100.0
return round(pct, 2)
def get_multiplier(self, pe_percentile: float) -> float:
"""
根据 PE 分位数返回对应的买入倍率
"""
for bracket in self.brackets:
if pe_percentile <= bracket['max_pct']:
return bracket['multiplier']
return self.brackets[-1]['multiplier']
"src/dca_engine.py"（★ 核心模块）
"""
dca_engine.py
★ 指数定投引擎：根据估值分位动态调整买入金额
"""
import pandas as pd
import numpy as np
from typing import Optional
class DCAEngine:
"""
通用指数定投机器人（支持估值分位加仓）
"""
def __init__(
self,
base_amount: float,
valuation_engine: "ValuationEngine",
frequency: str = "W"
):
"""
参数:
base_amount: 基准定投金额（元）
valuation_engine: 估值引擎实例
frequency: 定投频率，"W"=每周，"M"=每月
"""
self.base_amount = base_amount
self.ve = valuation_engine
self.frequency = frequency
# 状态记录
self.cash = 0.0 # 持有现金（元）
self.shares = 0.0 # 持有份额
self.total_invested = 0.0 # 累计投入
self.trade_log: list[dict] = [] # 交易日志
def _get_frequency_date(self, date: pd.Timestamp) -> bool:
"""
判断当前日期是否应该执行定投：
- W: 每周一
- M: 每月第一个交易日
"""
if self.frequency == "W":
return date.dayofweek == 0
elif self.frequency == "M":
return date.day == 1
return False
def run_daily(
self,
date: pd.Timestamp,
index_price: float,
pe_value: float,
pe_history: pd.Series
):
"""
每日调用，由回测引擎驱动
参数:
date: 当前日期
index_price: 当日指数点位（或 ETF 价格）
pe_value: 当日 PE
pe_history: 截至当日的历史 PE 序列
"""
# 只在定投日执行
if not self._get_frequency_date(date):
return
# ★ 核心：计算 PE 分位 + 获取买入倍率
pe_pct = self.ve.compute_percentile(pe_history, pe_value)
multiplier = self.ve.get_multiplier(pe_pct)
# 计算当期买入金额
invest_amount = self.base_amount * multiplier
if invest_amount <= 0:
# 暂停定投
self.trade_log.append({
'date': date,
'action': 'skip',
'reason': f'PE分位 {pe_pct:.1f}% → 倍率 {multiplier}x',
'amount': 0,
'price': index_price,
'shares': 0
})
return
# 执行买入
shares_bought = invest_amount / index_price
self.cash -= invest_amount
self.shares += shares_bought
self.total_invested += invest_amount
self.trade_log.append({
'date': date,
'action': 'buy',
'reason': f'PE分位 {pe_pct:.1f}% → 倍率 {multiplier}x',
'amount': invest_amount,
'price': index_price,
'shares': shares_bought
})
def get_snapshot(self, current_price: float) -> dict:
"""返回当前账户快照"""
market_value = self.shares * current_price
return {
'cash': self.cash,
'shares': self.shares,
'market_value': market_value,
'total_invested': self.total_invested,
'pnl': market_value - self.total_invested,
'pnl_pct': (
(market_value - self.total_invested) / self.total_invested * 100
if self.total_invested > 0 else 0
)
}
"src/backtester.py"
"""
backtester.py
回测引擎：驱动定投机器人
"""
import pandas as pd
from src.dca_engine import DCAEngine
def run_backtest(
engine: DCAEngine,
price_data: pd.DataFrame,
pe_data: pd.DataFrame,
start_date: str,
end_date: str
) -> dict:
"""
遍历历史数据，驱动定投引擎
"""
dates = price_data.index
if start_date:
dates = dates[dates >= pd.Timestamp(start_date)]
if end_date:
dates = dates[dates <= pd.Timestamp(end_date)]
for date in dates:
if date not in price_data.index:
continue
price = price_data.loc[date, 'close']
pe_val = pe_data.loc[date, 'pe'] if date in pe_data.index else None
if price <= 0 or pe_val is None:
continue
# 构建截至当前的历史 PE 序列
hist_pe = pe_data.loc[:date, 'pe']
engine.run_daily(date, price, pe_val, hist_pe)
return {
'trade_log': engine.trade_log,
'engine': engine
}
"main.py"
"""
main.py
主入口：运行估值分位加仓定投回测
"""
import yaml
import pandas as pd
from pathlib import Path
from src.data_loader import load_pe_history
from src.valuation import ValuationEngine
from src.dca_engine import DCAEngine
from src.backtester import run_backtest
def load_config(path='config.yaml'):
with open(path) as f:
return yaml.safe_load(f)
def generate_mock_data(n_days=1800):
"""生成模拟指数价格和 PE 数据"""
import numpy as np
dates = pd.date_range('2018-01-01', periods=n_days, freq='B')
np.random.seed(42)
# 模拟指数价格
price = pd.DataFrame({
'close': 3000 * pd.Series(
np.cumprod(1 + np.random.normal(0.0003, 0.015, n_days))
).values
}, index=dates)
# 模拟 PE：均值回归
pe_base = 12 + np.random.normal(0, 0.3, n_days)
pe_base = pd.Series(pe_base.cumsum() / range(1, n_days + 1) * 100, index=dates)
pe_df = pd.DataFrame({'pe': pe_base})
return price, pe_df
def main():
cfg = load_config()
# 加载/生成数据
try:
pe_data = load_pe_history('data/index_pe_history.csv')
price_data = pd.DataFrame({'close': pe_data['pe'] * 100})
except FileNotFoundError:
print("未找到数据文件，使用模拟数据")
price_data, pe_data = generate_mock_data()
# 初始化估值引擎
ve = ValuationEngine(cfg['valuation_brackets'])
# 初始化定投引擎
engine = DCAEngine(
base_amount=cfg['dca']['base_amount'],
valuation_engine=ve,
frequency=cfg['dca']['frequency']
)
# 运行回测
result = run_backtest(
engine, price_data, pe_data,
cfg['backtest']['start_date'],
cfg['backtest']['end_date']
)
# 打印结果
log = result['trade_log']
buys = [t for t in log if t['action'] == 'buy']
skips = [t for t in log if t['action'] == 'skip']
print(f"\n{'='*60}")
print(f" 指数定投回测结果")
print(f"{'='*60}")
print(f" 定投次数: {len(buys)} 次")
print(f" 暂停次数: {len(skips)} 次")
print(f" 累计投入: ¥{engine.total_invested:,.0f}")
snap = engine.get_snapshot(price_data['close'].iloc[-1])
print(f" 持有份额: {snap['shares']:.2f}")
print(f" 市值: ¥{snap['market_value']:,.2f}")
print(f" 累计收益: ¥{snap['pnl']:,.2f}")
print(f" 累计收益率: {snap['pnl_pct']:.2f}%")
print(f"{'='*60}\n")
# 打印前 10 条交易
print("前 10 条交易记录:")
for t in log[:10]:
print(f" {t['date'].strftime('%Y-%m-%d')} | "
f"{t['action']:>4} | "
f"金额 ¥{t['amount']:>8,.0f} | "
f"理由: {t['reason']}")
if __name__ == '__main__':
main()
六、README.md 与使用说明
# 通用指数定投机器人 — 估值分位加仓版
## 核心功能
根据指数 PE 历史分位数，自定义 5 档加仓梯度，自动调整每期定投金额。
## 安装
```bash
pip install -r requirements.txt
```
## 快速开始
### 1. 准备数据
创建 `data/index_pe_history.csv`：
```csv
date,pe
2018-01-02,12.5
2018-01-03,12.3
...
```
### 2. 配置五档梯度
编辑 `config.yaml`：
```yaml
dca:
base_amount: 1000 # 基准定投 1000 元/期
frequency: "W" # 每周一定投
valuation_brackets:
- { max_pct: 20, multiplier: 2.0 } # 很度低估 → 2000 元
- { max_pct: 40, multiplier: 1.5 } # 低估 → 1500 元
- { max_pct: 60, multiplier: 1.0 } # 正常 → 1000 元
- { max_pct: 80, multiplier: 0.5 } # 高估 → 500 元
- { max_pct: 100, multiplier: 0.0 } # 极度高估 → 暂停
```
### 3. 运行
```bash
python main.py
```
输出：定投次数、暂停次数、累计收益、交易明细
## 自定义档位
你可以自由修改 `valuation_brackets`，例如：
| 风格 | 档位设置 |
|------|----------|
| 激进型 | 低估 3x、正常 1.5x、高估 0.5x |
| 保守型 | 低估 1.5x、正常 1x、高估 0x |
| 自定义 | 任意分位 + 任意倍率 |
七、核心知识点卡片
┌──────────────────────────────────────────────────────────────┐
│ 指数定投 + 估值分位加仓 — 核心知识 │
├────────────────┬─────────────────────────────────────────────┤
│ 估值分位 │ 当前 PE 在历史上的百分位（0~100%） │
│ 分位越低 │ 越便宜，越应加大买入 │
│ 五档梯度 │ 自定义分位区间 + 对应买入倍率 │
│ 基准金额 │ 固定锚点，所有倍率基于它计算 │
│ 暂停定投 │ 高位（如 ≥80% 分位）时倍率 0x │
│ 核心优势 │ 便宜多买、贵时少买，摊薄长期成本 │
│ 适用标的 │ 宽基指数（沪深 300/中证 500/创业板指） │
│ 频率选择 │ 周定投平滑波动，月定投减少手续费冲击 │
└────────────────┴─────────────────────────────────────────────┘
八、免责声明与风险提示
⚠️ 免责声明：本代码仅供学习、研究与量化教学用途，不构成任何投资建议或投资决策依据。模拟数据为随机数生成，不代表任何真实指数或标的历史表现。
⚠️ 风险提示：
- 历史 PE 分位不能完全预测未来，市场估值中枢可能长期抬升或下移
- 极端行情下 PE 可能失真（如盈利大幅波动时）
- 定投频率与费率结构会显著影响实际收益
- 回测结果不代表实盘表现，未考虑滑点、申赎费率差异等
九、总结
给通用指数定投机器人加上 估值分位五档加仓梯度，核心价值在于：
1. 便宜多买、贵时少买 —— 用历史估值做锚，系统化提升定投性价比
2. 五档梯度完全自定义 —— 激进/保守/个性化策略都能灵活配置
3. 逻辑清晰、可回测验证 —— 每一笔买入都有明确的估值依据
4. 适合宽基指数长期投资 —— 沪深 300、中证 500 等流动性好、估值稳定的标的
核心原则：定投的本质是"分批入场、摊薄成本"，加上估值分位判断后，它变成了"聪明地分批入场"——这正是指数投资的长期致胜之道。
本文代码仅供学习和技术交流，不构成任何投资建议，股市有风险，入市需谨慎！
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！