好，我直接给你一版适合你现在这套系统的完整项目结构，目标就是把你现在的： 策略计算 Excel 输出 Word 报告 实盘执行回填 偏差跟踪 全部放进一个清晰、能长期维护的架构里。你现在这份日报已经有“大势、行业、今日交易、当前持仓、明日建议”这些基础块，适合作为这个架构里的报告层输出。 lArPD1q48E0hVonOGHh83M4zRCVv 一、推荐目录结构 quant_system/ ├── main.py ├── config.py ├── requirements.txt │ ├── data/ │ ├── raw/ │ ├── processed/ │ └── live/ │ ├── strategy/ │ ├── __init__.py │ ├── regime_engine.py │ ├── alpha_engine.py │ ├── portfolio_constructor.py │ ├── execution_engine.py │ └── backtest_engine.py │ ├── execution/ │ ├── __init__.py │ ├── planned_orders.py │ ├── execution_log.py │ ├── execution_diff.py │ └── broker_import.py │ ├── reporting/ │ ├── __init__.py │ ├── excel_exporter.py │ ├── word_reporter.py │ └── report_templates.py │ ├── storage/ │ ├── __init__.py │ ├── snapshot_store.py │ ├── trade_store.py │ └── position_store.py │ ├── outputs/ │ ├── excel/ │ ├── word/ │ └── logs/ │ └── utils/ ├── __init__.py ├── dates.py ├── formatters.py └── helpers.py 二、每个模块负责什么 strategy/ 只负责策略本身，不碰 Word 和 Excel。 regime_engine.py 大势判断，输出状态、目标仓位 alpha_engine.py 选股、打分、排名 portfolio_constructor.py 目标持仓、目标权重 execution_engine.py 理论调仓动作 backtest_engine.py 回测专用 execution/ 只负责“程序建议”和“你实际执行”的闭环。 planned_orders.py 生成明日执行清单 execution_log.py 读取/维护你的实际成交记录 execution_diff.py 比较程序建议 vs 实际执行 reporting/ 只负责输出。 excel_exporter.py 导出各类表格 word_reporter.py 生成每日 Word 报告 report_templates.py 放报告标题、字段顺序、固定提示语 storage/ 只负责存档。 snapshot_store.py 每日策略摘要 trade_store.py 计划订单、执行记录、偏差表 position_store.py 当前持仓、成本价、历史变动 三、每天的实际流程 你的系统以后每天固定走这条链： 收盘后更新数据 → 策略模块跑出 regime / selected / target_weights → execution 模块生成 planned_orders → reporting 导出 Excel → reporting 生成 Word 执行报告 → 你第二天执行 → 把实际成交填入 execution_log → 程序自动生成 execution_diff → 下一天报告自动显示执行偏差 四、主程序 main.py 应该怎么写 下面这版是你可以直接照着搭的主流程骨架： from strategy.regime_engine import RegimeEngine from strategy.alpha_engine import AlphaEngine from strategy.portfolio_constructor import PortfolioConstructor from execution.planned_orders import build_planned_orders from execution.execution_log import load_or_init_execution_log from execution.execution_diff import build_execution_diff from reporting.excel_exporter import export_daily_package from reporting.word_reporter import generate_word_report from storage.snapshot_store import build_strategy_snapshot from storage.position_store import build_holdings_table def run_daily_process(data, report_date, next_trade_date, positions, entry_costs, total_value): regime_engine = RegimeEngine() alpha_engine = AlphaEngine() portfolio_constructor = PortfolioConstructor() regime_info = regime_engine.get_regime_info( date=report_date, data=data, prev_state=data.get("prev_state"), prev_score=data.get("prev_score"), consecutive_days=data.get("consecutive_days", 0), prev_pos=data.get("prev_pos", 0), ) final_state = regime_info["final_state"] target_exposure = regime_info["target_exposure"] selected, rank_map = alpha_engine.select_stocks( data=data, date=report_date, regime=final_state, current_positions=positions ) target_weights = portfolio_constructor.calculate_target_weights( selected=selected, regime=final_state ) planned_orders_df = build_planned_orders( date=report_date, selected=selected, positions=positions, target_weights=target_weights, total_value=total_value, target_exposure=target_exposure, final_state=final_state, price_df=data["price"], rank_map=rank_map ) holdings_df = build_holdings_table( date=report_date, positions=positions, price_df=data["price"], entry_costs=entry_costs, rank_map=rank_map ) snapshot_df = build_strategy_snapshot( date=report_date, strategy_version="v2.0", regime_info=regime_info, total_value=total_value, actual_exposure=sum( positions.get(code, 0) * float(data["price"].loc[report_date, code]) for code in positions if code in data["price"].columns ) / max(total_value, 1), hold_count=len(positions), is_rebalance_day=False, summary=f"当前 {final_state}，目标仓位 {target_exposure:.0%}" ) execution_log_df = load_or_init_execution_log( plan_date=report_date, next_trade_date=next_trade_date, planned_orders_df=planned_orders_df ) execution_diff_df = build_execution_diff( planned_orders_df=planned_orders_df, execution_log_df=execution_log_df ) export_daily_package( report_date=report_date, snapshot_df=snapshot_df, holdings_df=holdings_df, planned_orders_df=planned_orders_df, execution_log_df=execution_log_df, execution_diff_df=execution_diff_df ) generate_word_report( report_date=report_date, snapshot_df=snapshot_df, holdings_df=holdings_df, planned_orders_df=planned_orders_df, execution_log_df=execution_log_df, execution_diff_df=execution_diff_df ) 五、几个核心文件的最小实现 1）execution/planned_orders.py import pandas as pd def attach_execution_zone(ref_price, buy_pct=0.02, watch_pct=0.04): if ref_price <= 0: return None, None, None, None return ( round(ref_price, 2), round(ref_price * (1 + buy_pct), 2), round(ref_price * (1 + watch_pct), 2), round(ref_price * (1 + watch_pct), 2), ) def build_planned_orders(date, selected, positions, target_weights, total_value, target_exposure, final_state, price_df, rank_map=None): rank_map = rank_map or {} rows = [] selected_codes = [x[0] for x in selected] target_total_hold = total_value * target_exposure target_shares = {} for code, price, score in selected: desired_value = target_total_hold * target_weights.get(code, 0.0) shares = int(desired_value / max(price, 0.01) / 100) * 100 target_shares[code] = shares for code, current_shares in positions.items(): ref_price = float(price_df.loc[date, code]) if code not in selected_codes: rows.append({ "date": date, "code": code, "action": "sell", "priority": "high", "current_shares": current_shares, "target_shares": 0, "share_change": -current_shares, "ref_price": ref_price, "buy_zone_low": None, "buy_zone_high": None, "watch_zone_high": None, "cancel_above": None, "rank": rank_map.get(code), "reason": "跌出目标池", "regime": final_state }) else: diff = target_shares[code] - current_shares if diff < 0: rows.append({ "date": date, "code": code, "action": "reduce", "priority": "high", "current_shares": current_shares, "target_shares": target_shares[code], "share_change": diff, "ref_price": ref_price, "buy_zone_low": None, "buy_zone_high": None, "watch_zone_high": None, "cancel_above": None, "rank": rank_map.get(code), "reason": "目标权重下降或风险降仓", "regime": final_state }) for code, price, score in selected: current_shares = positions.get(code, 0) diff = target_shares[code] - current_shares if diff > 0: low, high, watch, cancel = attach_execution_zone(price) rows.append({ "date": date, "code": code, "action": "buy", "priority": "medium", "current_shares": current_shares, "target_shares": target_shares[code], "share_change": diff, "ref_price": round(price, 2), "buy_zone_low": low, "buy_zone_high": high, "watch_zone_high": watch, "cancel_above": cancel, "rank": rank_map.get(code), "reason": "进入目标池", "regime": final_state }) return pd.DataFrame(rows) 2）execution/execution_log.py from pathlib import Path import pandas as pd def load_or_init_execution_log(plan_date, next_trade_date, planned_orders_df): path = Path(f"outputs/excel/execution_log_{plan_date}.xlsx") if path.exists(): return pd.read_excel(path) rows = [] for _, r in planned_orders_df.iterrows(): if r["action"] == "hold": continue rows.append({ "trade_date": next_trade_date, "plan_date": plan_date, "code": r["code"], "action_planned": r["action"], "action_executed": "", "planned_shares": r["share_change"], "filled_shares": 0, "filled_price": None, "status": "", "deviation_reason": "", "note": "" }) return pd.DataFrame(rows) 3）execution/execution_diff.py import pandas as pd def build_execution_diff(planned_orders_df, execution_log_df): merged = planned_orders_df.merge( execution_log_df, how="left", left_on=["date", "code", "action"], right_on=["plan_date", "code", "action_planned"] ) rows = [] for _, r in merged.iterrows(): planned = abs(int(r["share_change"])) filled = abs(int(r["filled_shares"])) if pd.notna(r["filled_shares"]) else 0 completion = filled / planned if planned > 0 else 1.0 ref_price = float(r["ref_price"]) if pd.notna(r["ref_price"]) else 0 filled_price = float(r["filled_price"]) if pd.notna(r["filled_price"]) else 0 slippage = None if ref_price > 0 and filled_price > 0: if r["action"] == "buy": slippage = round((filled_price - ref_price) / ref_price * 100, 2) else: slippage = round((ref_price - filled_price) / ref_price * 100, 2) rows.append({ "date": r["date"], "code": r["code"], "planned_action": r["action"], "executed_action": r.get("action_executed", ""), "planned_shares": r["share_change"], "filled_shares": r.get("filled_shares", 0), "completion_ratio": round(completion, 4), "ref_price": ref_price, "filled_price": filled_price if filled_price > 0 else None, "slippage_pct": slippage, "status": r.get("status", ""), "deviation_reason": r.get("deviation_reason", "") }) return pd.DataFrame(rows) 4）reporting/excel_exporter.py from pathlib import Path def export_daily_package(report_date, snapshot_df, holdings_df, planned_orders_df, execution_log_df, execution_diff_df): out = Path("outputs/excel") out.mkdir(parents=True, exist_ok=True) snapshot_df.to_excel(out / f"strategy_snapshot_{report_date}.xlsx", index=False) holdings_df.to_excel(out / f"holdings_{report_date}.xlsx", index=False) planned_orders_df.to_excel(out / f"planned_orders_{report_date}.xlsx", index=False) execution_log_df.to_excel(out / f"execution_log_{report_date}.xlsx", index=False) execution_diff_df.to_excel(out / f"execution_diff_{report_date}.xlsx", index=False) 5）reporting/word_reporter.py from pathlib import Path from docx import Document def add_df_table(doc, title, df): doc.add_heading(title, level=1) if df.empty: doc.add_paragraph("无数据") return table = doc.add_table(rows=1, cols=len(df.columns)) table.style = "Table Grid" for i, col in enumerate(df.columns): table.rows[0].cells[i].text = str(col) for _, row in df.iterrows(): cells = table.add_row().cells for i, col in enumerate(df.columns): cells[i].text = "" if row[col] is None else str(row[col]) def generate_word_report(report_date, snapshot_df, holdings_df, planned_orders_df, execution_log_df, execution_diff_df): out = Path("outputs/word") out.mkdir(parents=True, exist_ok=True) doc = Document() doc.add_heading("每日量化执行报告", 0) s = snapshot_df.iloc[0] doc.add_paragraph(f"报告日期: {report_date}") doc.add_paragraph(f"大势状态: {s['regime']}") doc.add_paragraph(f"大势评分: {s['regime_score']}") doc.add_paragraph(f"目标仓位: {s['target_position']:.0%}") doc.add_paragraph(f"实际仓位: {s['actual_position']:.0%}") doc.add_paragraph(f"净值: {s['nav']:,.2f}") doc.add_paragraph(f"持仓数: {s['hold_count']}") add_df_table(doc, "当前持仓分析", holdings_df) add_df_table(doc, "明日执行清单", planned_orders_df) add_df_table(doc, "今日执行记录", execution_log_df) add_df_table(doc, "执行偏差分析", execution_diff_df) doc.add_heading("风险提示", level=1) doc.add_paragraph("卖出和减仓优先于买入执行。") doc.add_paragraph("买入请按参考价和区间执行，高开过多不追。") doc.add_paragraph("若连续多日未成交，请重新评估信号而非死守旧价格。") doc.save(out / f"量化执行报告_{report_date}.docx") 六、你现有 v2.0 最值得立刻补的两个点 1）成本价记录 现在你 Word 里“当前持仓”只有股数、现价、建议。 最好补： 成本价 浮盈亏% 当前排名 理由 这样报告会从“看仓位”升级成“看持仓质量”。 2）明日执行清单 你现在的 Word 更像日报，不像执行单。 必须加： 优先级 股数变化 参考价 买入区间 放弃条件 否则第二天还是不够直接。 七、我对你这套系统的最终建议 现在不要再优先花时间调策略参数。 先把这个闭环搭好： 计划订单 → 实际执行 → 偏差统计 → Word 报告 这一步做完，你的系统就从“会回测的程序”升级成“能长期跑实盘流程的程序”。