13_Tech_Python_Engineering

Python 工程实践

本项目不是"想到哪写到哪"的数据分析脚本，而是按生产级标准设计的系统。本篇总结核心工程原则。

---

1. 计算与 IO 完全分离

这是整个系统最重要的设计决策。

• core/wyckoff_engine.py 是纯计算引擎——没有任何网络请求，只接收 pandas.DataFrame 输入，只做数学和逻辑运算。
• 网络请求（拉行情、调 API）由外层入口负责（如 scripts/wyckoff_funnel.py）。

为什么这样做：因为计算和 IO 解耦后，可以把行情数据存为快照文件，然后离线回放。调参数、跑回测、排查 bug 都不需要连网。

---

2. 配置与逻辑分离

• FunnelConfig（core/wyckoff_engine.py）用 dataclass 定义所有阈值参数
• 不在代码里写魔法数字
• 大盘水温系统可以在运行前修改 FunnelConfig 的属性来动态调节全局阈值

这意味着：同一份代码 + 不同配置 = 不同策略风格，无需改代码。

---

3. 类型注解

核心模块（core/、scripts/）使用严格的类型注解：

def detect_sos(df: pd.DataFrame, cfg: FunnelConfig) -> float | None:

好处：

• 降低 DataFrame 处理中 NaN/None 的隐性 bug
• IDE 悬停即可看到完整签名和字段补全

---

4. 容错降级

金融定时任务不能因为一个次要模块崩溃就整条链路挂掉：

场景	降级策略
tushare 取数失败	自动切 akshare → baostock → efinance
RAG 新闻接口崩溃	跳过防雷，记录 Warning，研报继续生成
合规简报模型失败	降级为本地模板
Step4 缺少 Supabase 配置	按设计跳过，不报错

---

5. 向量化优先

涉及全市场横向比较的计算（比如 RPS 全市场排名）全部用 Pandas 向量化操作：

rps_df["ret_fast"].rank(pct=True)

不用 for 循环逐只股票计算。这让 ~4500 只股票的每日扫描在秒级完成。

---

6. 扁平架构

遵循 KISS 原则：

• 没有微服务、没有消息队列
• 没有自定义 ORM
• 用 GitHub Actions 做调度，不需要独立的任务引擎
• 用 SQLite 做本地存储，不需要数据库服务器

够用就不加复杂度。