策略每日运行日志自动存档程序（留存策略参数与操作记录）｜教学级量化交易原型
内容包含免责声明和风险提示，不荐股、不自动化交易、不引导开户、无任何引流。
一、实际应用场景描述
在智能证券投资课程中，策略运行日志（Strategy Run Log）与可复现性（Reproducibility）是量化交易系统开发的核心教学内容。
本程序适用于：
- 高校量化投资、算法交易课程实验
- 个人量化策略开发与调试
- 策略运行审计与归因分析教学
- 多策略并行管理的日志系统演示
核心目标：
- 每日自动记录策略运行参数
- 留存当日操作记录（信号、下单、持仓变更）
- 结构化存档，便于后续复盘与审计
- 强调本地存储、用户自主管理
✅ 不连接券商系统
✅ 不执行真实交易
✅ 仅作为策略日志管理教学工具
二、痛点引入（真实可感知）
痛点 表现
"那天为啥买？忘了" 缺乏操作记录，复盘无据可查
参数漂移 今天改了参数，明天不记得了
无法复现 历史回测和实盘差异无法定位
多策略混乱 同时跑 3 个策略，日志混在一起
工具过重 专业日志系统（ELK 等）对个人过重
👉 需要一个轻量、本地、可解释、可复现的策略日志工具
三、核心逻辑讲解（工程视角）
1️⃣ 数据模型设计
StrategyRunLog
├── run_id 运行唯一 ID（UUID）
├── date 运行日期
├── strategy_name 策略名称
├── parameters 策略参数快照（JSON）
├── signals 当日产生的信号列表
├── orders 当日操作记录列表
└── notes 备注（如 市场异常、临时调整）
2️⃣ 日志内容设计（教学用）
模块 记录内容
策略参数 均线周期、RSI 阈值、仓位上限等
市场快照 当日行情概要（开盘 / 收盘 / 波动率）
信号记录 买入 / 卖出 / 观望 + 触发条件
操作记录 下单方向、数量、价格（模拟）
异常标记 数据缺失、信号冲突等
3️⃣ 存档流程
策略每日运行结束
↓
收集：参数 + 信号 + 操作
↓
生成唯一 run_id
↓
序列化为 JSON
↓
按日期目录存储
4️⃣ 目录结构设计
strategy_logs/
├── 2025-07-14/
│ ├── run_1a2b3c.json
│ └── run_4d5e6f.json
├── 2025-07-15/
│ └── run_7g8h9i.json
└── index.json # 总索引
5️⃣ 设计原则
- 只记录，不执行
- 参数快照化，确保可复现
- 每日独立存档，便于审计
四、Python 模块化代码（可直接运行）
📁 项目结构
strategy_logger/
│
├── main.py
├── models.py
├── logger.py
├── serializer.py
├── reporter.py
├── storage.py
├── README.md
└── DISCLAIMER.md
✅ models.py（数据建模）
"""
models.py
策略运行日志数据模型
"""
import uuid
import json
from datetime import datetime
class StrategyRunLog:
"""单次策略运行日志"""
def __init__(
self,
strategy_name,
parameters,
market_snapshot,
signals=None,
orders=None,
notes=""
):
"""
strategy_name: 策略名称
parameters: 策略参数字典
market_snapshot: 市场快照字典
signals: 信号列表
orders: 操作记录列表
notes: 备注
"""
self.run_id = str(uuid.uuid4())
self.date = datetime.now().strftime("%Y-%m-%d")
self.timestamp = datetime.now().isoformat()
self.strategy_name = strategy_name
self.parameters = parameters
self.market_snapshot = market_snapshot
self.signals = signals or []
self.orders = orders or []
self.notes = notes
def to_dict(self):
"""序列化为字典"""
return {
"run_id": self.run_id,
"date": self.date,
"timestamp": self.timestamp,
"strategy_name": self.strategy_name,
"parameters": self.parameters,
"market_snapshot": self.market_snapshot,
"signals": self.signals,
"orders": self.orders,
"notes": self.notes
}
@classmethod
def from_dict(cls, data):
"""从字典还原"""
log = cls(
data["strategy_name"],
data["parameters"],
data["market_snapshot"],
data["signals"],
data["orders"],
data["notes"]
)
log.run_id = data["run_id"]
log.date = data["date"]
log.timestamp = data["timestamp"]
return log
✅ logger.py（核心日志引擎）
"""
logger.py
策略运行日志引擎
"""
import os
import json
from datetime import datetime
from models import StrategyRunLog
class StrategyLogger:
"""策略日志管理器"""
def __init__(self, base_dir="strategy_logs"):
self.base_dir = base_dir
os.makedirs(base_dir, exist_ok=True)
self.index_path = os.path.join(base_dir, "index.json")
self._load_index()
def _load_index(self):
"""加载总索引"""
if os.path.exists(self.index_path):
with open(self.index_path, "r") as f:
self.index = json.load(f)
else:
self.index = {"total_runs": 0, "runs": []}
def _save_index(self):
"""保存总索引"""
with open(self.index_path, "w") as f:
json.dump(self.index, f, ensure_ascii=False, indent=2)
def log_run(self, run_log):
"""
核心方法：存档一次策略运行记录
"""
# 按日期创建子目录
date_dir = os.path.join(self.base_dir, run_log.date)
os.makedirs(date_dir, exist_ok=True)
# 写入当日运行记录
filename = f"run_{run_log.run_id[:8]}.json"
filepath = os.path.join(date_dir, filename)
with open(filepath, "w") as f:
json.dump(run_log.to_dict(), f, ensure_ascii=False, indent=2)
# 更新总索引
self.index["total_runs"] += 1
self.index["runs"].append({
"run_id": run_log.run_id,
"date": run_log.date,
"strategy": run_log.strategy_name,
"file": os.path.join(run_log.date, filename),
"signal_count": len(run_log.signals),
"order_count": len(run_log.orders)
})
self._save_index()
return filepath
def get_run(self, run_id):
"""按 run_id 检索"""
for entry in self.index["runs"]:
if entry["run_id"] == run_id:
filepath = entry["file"]
with open(os.path.join(self.base_dir, filepath), "r") as f:
return StrategyRunLog.from_dict(json.load(f))
return None
def list_runs(self, date=None, strategy_name=None):
"""列出所有运行记录，可选过滤"""
results = self.index["runs"]
if date:
results = [r for r in results if r["date"] == date]
if strategy_name:
results = [r for r in results if r["strategy"] == strategy_name]
return results
✅ serializer.py（参数快照工具）
"""
serializer.py
策略参数快照与对比工具
"""
def snapshot_parameters(params_dict):
"""将策略参数固化为可复现的快照"""
import json
return json.dumps(params_dict, sort_keys=True, ensure_ascii=False)
def compare_parameters(snap_a, snap_b):
"""对比两次运行的参数差异"""
import json
a = json.loads(snap_a)
b = json.loads(snap_b)
diffs = {}
all_keys = set(list(a.keys()) + list(b.keys()))
for k in all_keys:
if a.get(k) != b.get(k):
diffs[k] = {"old": a.get(k), "new": b.get(k)}
return diffs
✅ reporter.py（日志报告输出）
"""
reporter.py
策略运行日志报告
"""
def report_run(run_log):
"""打印单次运行日志"""
print("\n" + "=" * 65)
print(f"【策略运行日志】")
print("=" * 65)
print(f" Run ID：{run_log.run_id}")
print(f" 策略：{run_log.strategy_name}")
print(f" 日期：{run_log.date}")
print(f" 时间：{run_log.timestamp}")
print(f"\n📋 策略参数快照：")
print("-" * 65)
for k, v in run_log.parameters.items():
print(f" {k} = {v}")
print(f"\n📊 市场快照：")
print("-" * 65)
for k, v in run_log.market_snapshot.items():
print(f" {k} = {v}")
if run_log.signals:
print(f"\n📡 信号记录（{len(run_log.signals)} 条）：")
print("-" * 65)
for i, sig in enumerate(run_log.signals, 1):
print(f" {i}. [{sig.get('time', 'N/A')}] {sig.get('type', '')}")
print(f" 条件：{sig.get('condition', '')}")
print(f" 结果：{sig.get('result', '')}")
if run_log.orders:
print(f"\n📝 操作记录（{len(run_log.orders)} 条）：")
print("-" * 65)
for i, order in enumerate(run_log.orders, 1):
print(f" {i}. {order.get('action', '')} | "
f"数量：{order.get('qty', '')} | "
f"价格：{order.get('price', '')}")
if run_log.notes:
print(f"\n📌 备注：{run_log.notes}")
print("=" * 65)
def report_index(logger):
"""打印总索引"""
print(f"\n📂 日志总览：共 {logger.index['total_runs']} 次运行")
print("-" * 65)
for r in logger.index["runs"]:
print(f" {r['date']} | {r['strategy']} | "
f"信号 {r['signal_count']} 条 | "
f"操作 {r['order_count']} 条")
✅ storage.py（本地存储——兼容性封装）
"""
storage.py
兼容性封装：委托给 logger 模块
"""
from logger import StrategyLogger
# 暴露给 main.py 使用
__all__ = ["StrategyLogger"]
✅ main.py（交互入口）
"""
main.py
策略每日运行日志自动存档工具
"""
from datetime import datetime
from models import StrategyRunLog
from logger import StrategyLogger
from serializer import snapshot_parameters
from reporter import report_run, report_index
def main():
print("=== 策略运行日志自动存档工具（教学版）===")
print("留存每一日策略运行参数与操作记录\n")
logger = StrategyLogger()
while True:
print(f"\n📋 选择操作：")
print(" 1. 存档今日策略运行")
print(" 2. 查看运行记录列表")
print(" 3. 查看某次运行详情")
print(" 4. 退出")
choice = input("\n请选择（1/2/3/4）：")
if choice == "1":
# 存档
strategy_name = input("策略名称：")
print("\n📌 策略参数（JSON 格式，空行结束）：")
print("示例：ma_short: 5")
params = {}
while True:
line = input()
if not line:
break
k, v = line.split(":", 1)
# 尝试转为数字
v = v.strip()
try:
v = float(v) if "." in v else int(v)
except ValueError:
pass
params[k.strip()] = v
print("\n📌 市场快照（JSON 格式，空行结束）：")
print("示例：close: 3250.8")
market = {}
while True:
line = input()
if not line:
break
k, v = line.split(":", 1)
v = v.strip()
try:
v = float(v) if "." in v else int(v)
except ValueError:
pass
market[k.strip()] = v
# 信号录入
signals = []
print("\n📡 录入信号（空行结束）：")
while True:
t = input("信号类型（如 BUY/SSELL/HOLD，空结束）：")
if not t:
break
cond = input("触发条件：")
res = input("信号结果：")
time_str = input("触发时间（HH:MM，可选）：") or "N/A"
signals.append({
"type": t,
"condition": cond,
"result": res,
"time": time_str
})
# 操作录入
orders = []
print("\n📝 录入操作（空行结束）：")
while True:
act = input("操作（BUY/SELL，空结束）：")
if not act:
break
qty = int(input("数量："))
price = float(input("价格："))
orders.append({
"action": act,
"qty": qty,
"price": price
})
notes = input("\n📌 备注（可选）：")
# 创建日志
run_log = StrategyRunLog(
strategy_name, params, market, signals, orders, notes
)
# 存档
filepath = logger.log_run(run_log)
print(f"\n✅ 日志已存档：{filepath}")
print(f" Run ID：{run_log.run_id}")
elif choice == "2":
report_index(logger)
elif choice == "3":
run_id = input("输入 Run ID（完整或前 8 位）：")
# 模糊匹配
matched = None
for r in logger.index["runs"]:
if r["run_id"].startswith(run_id):
matched = logger.get_run(r["run_id"])
break
if matched:
report_run(matched)
else:
print("⚠️ 未找到对应 Run ID")
elif choice == "4":
print("\n👋 再见！日志已安全存档。")
break
else:
print("⚠️ 无效选择")
if __name__ == "__main__":
main()
五、README 与使用说明
# 策略每日运行日志自动存档工具（教学版）
## 项目说明
留存每一日策略运行参数与操作记录，便于复盘与审计。
## 使用方式
```bash
python main.py
```
## 操作流程
### 存档运行记录
```
选择操作：1
策略名称：均线突破策略
ma_short: 5
ma_long: 20
rsi_threshold: 30
（空行结束）
close: 3250.8
volume: 185000000
volatility: 1.8
（空行结束）
信号类型：BUY
触发条件：MA5 > MA20 and RSI < 30
信号结果：生成买入信号
触发时间：09:45
（空行结束，可录多条）
操作：BUY
数量：100
价格：3248.5
（空行结束，可录多条）
```
## 存储结构
```
strategy_logs/
├── 2025-07-14/
│ └── run_1a2b3c4d.json
└── index.json
```
## 适用范围
- 量化投资策略课程
- 策略开发与调试
- 运行审计与归因分析
## 注意事项
- 不执行真实交易
- 不连接券商系统
- 使用前请阅读 DISCLAIMER.md
六、DISCLAIMER.md（免责声明与风险提示）
# 免责声明与风险提示
## 免责声明
本程序仅供**教学与科研用途**，用于演示策略运行日志的归档方法。
作者不提供任何投资建议，不推荐任何策略，不承诺任何收益。
## 风险提示
1. 日志完整性依赖程序正常退出，异常中断可能丢失当日记录
2. JSON 存储不适合高频策略（每秒数百条信号），此类场景应使用时序数据库
3. Run ID 为随机 UUID，不支持自定义命名，查找需借助索引
4. 参数快照仅记录提交时的状态，运行时动态修改不会被捕获
5. 日志文件未加密，多用户环境需注意权限控制
使用本工具产生的任何后果，作者概不负责。
七、核心知识点卡片（教学向）
分类 内容
Python 类、UUID、JSON 序列化、文件系统操作
量化交易 策略可复现性、参数快照、运行审计
系统工程 日志分层设计、索引管理、按日分区
数据分析 历史回放、参数漂移检测
工程思想 模块化、可扩展、可审计
可扩展性 可接入数据库、支持多策略并行写入
八、总结（工程师视角）
这是一个完全中立、去营销化、可教学的原型系统：
✅ 不鼓吹任何策略
✅ 不伪装成日志平台
✅ 不替代专业运维工具
它真正展示的是：
如何用 Python 把"今天策略怎么跑的"从模糊记忆，变成可检索、可复现、可审计的结构化资产
核心教学价值：
传统做法 日志化后
"参数好像改过" 每日快照，精确对比
"为什么今天信号不一样" 参数 + 市场快照 → 可回放
"当时为什么买" 信号记录 + 触发条件 = 完整归因
"策略失效了" 逐日对比参数漂移，定位退化点
日志是量化交易者的"黑匣子"——没它，你飞了一万小时，连自己怎么飞的都不知道。
本文代码仅供学习和技术交流，不构成任何投资建议，股市有风险，入市需谨慎！
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
