“通用指数定投机器人” 本身是一个定期定额或定期不定额投资指数的策略现在要给它加上 估值分位加仓档位并且是 自定义的 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.txtpandas1.5numpy1.21pyyaml6.0matplotlib3.5config.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-01end_date: 2024-12-31src/data_loader.pydata_loader.py加载指数历史 PE 数据import pandas as pddef load_pe_history(filepath: str) - pd.DataFrame:加载指数历史 PE 数据预期格式:date,pe2018-01-02,12.52018-01-03,12.3...df pd.read_csv(filepath, parse_dates[date])df df.set_index(date).sort_index()return dfsrc/valuation.py★ 核心模块valuation.py★ 估值分位计算模块import pandas as pdimport numpy as npclass 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, keylambda x: x[max_pct])def compute_percentile(self, pe_series: pd.Series, current_pe: float) - float:计算当前 PE 在历史中的分位数0~100逻辑(历史 PE 当前 PE 的天数) / (总天数) × 100if len(pe_series) 0:return 50.0 # 无数据时默认中位count (pe_series current_pe).sum()pct (count / len(pe_series)) * 100.0return 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 pdimport numpy as npfrom typing import Optionalclass DCAEngine:通用指数定投机器人支持估值分位加仓def __init__(self,base_amount: float,valuation_engine: ValuationEngine,frequency: str W):参数:base_amount: 基准定投金额元valuation_engine: 估值引擎实例frequency: 定投频率W每周M每月self.base_amount base_amountself.ve valuation_engineself.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 0elif self.frequency M:return date.day 1return Falsedef run_daily(self,date: pd.Timestamp,index_price: float,pe_value: float,pe_history: pd.Series):每日调用由回测引擎驱动参数:date: 当前日期index_price: 当日指数点位或 ETF 价格pe_value: 当日 PEpe_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 * multiplierif invest_amount 0:# 暂停定投self.trade_log.append({date: date,action: skip,reason: fPE分位 {pe_pct:.1f}% → 倍率 {multiplier}x,amount: 0,price: index_price,shares: 0})return# 执行买入shares_bought invest_amount / index_priceself.cash - invest_amountself.shares shares_boughtself.total_invested invest_amountself.trade_log.append({date: date,action: buy,reason: fPE分位 {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_pricereturn {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 * 100if self.total_invested 0 else 0)}src/backtester.pybacktester.py回测引擎驱动定投机器人import pandas as pdfrom src.dca_engine import DCAEnginedef run_backtest(engine: DCAEngine,price_data: pd.DataFrame,pe_data: pd.DataFrame,start_date: str,end_date: str) - dict:遍历历史数据驱动定投引擎dates price_data.indexif 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:continueprice price_data.loc[date, close]pe_val pe_data.loc[date, pe] if date in pe_data.index else Noneif 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.pymain.py主入口运行估值分位加仓定投回测import yamlimport pandas as pdfrom pathlib import Pathfrom src.data_loader import load_pe_historyfrom src.valuation import ValuationEnginefrom src.dca_engine import DCAEnginefrom src.backtester import run_backtestdef load_config(pathconfig.yaml):with open(path) as f:return yaml.safe_load(f)def generate_mock_data(n_days1800):生成模拟指数价格和 PE 数据import numpy as npdates pd.date_range(2018-01-01, periodsn_days, freqB)np.random.seed(42)# 模拟指数价格price pd.DataFrame({close: 3000 * pd.Series(np.cumprod(1 np.random.normal(0.0003, 0.015, n_days))).values}, indexdates)# 模拟 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, indexdates)pe_df pd.DataFrame({pe: pe_base})return price, pe_dfdef 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_amountcfg[dca][base_amount],valuation_engineve,frequencycfg[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 档加仓梯度自动调整每期定投金额。## 安装bashpip install -r requirements.txt## 快速开始### 1. 准备数据创建 data/index_pe_history.csvcsvdate,pe2018-01-02,12.52018-01-03,12.3...### 2. 配置五档梯度编辑 config.yamlyamldca: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. 运行bashpython 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解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛