AI Agent量化交易自动化实战：Hermes Agent 20个任务架构与代码
量化投资中那些”不得不做”的重复劳动
做量化投资三年，我最深刻的体会是：策略研究只占20%的时间，剩下80%都在做”搬砖”。
每天的工作流是这样的：
08:30 检查昨日数据是否完整 → 缺数据？手动补
09:30 开盘前检查网关、磁盘、进程 → 出问题？手动重启
15:00 收盘后下载今日行情 → 跑脚本、等结果
16:00 生成日报 → 查SQL、画图表、写评论
17:00 更新信号监控 → 看预警、调整参数
21:00 复盘今日交易 → 翻日志、对账单
周五 跑周报 → 全市场估值、可转债筛选
周日 更新模型 → 重新训练、整理知识库这些工作不是不能自动化，而是很难做好自动化——每个环节都可能出异常：数据源断开、磁盘满了、代理挂了、限流了。传统cron脚本遇到异常就崩溃，第二天才发现少跑了一天的数据。
我需要的不只是定时执行，而是一个能感知上下文、能自动恢复、能主动通知我的自动化框架。
这就引出了本文的主角——Hermes Agent。
为什么选择Hermes Agent？它不是”加强版cron”
选型时我调研了以下几个方案：
| 方案 | 优点 | 缺点 |
|---|---|---|
| 原生cron | 简单 | 无错误处理、无上下文、无通知 |
| Airflow | 功能强大 | 太重了，个人量化杀鸡用牛刀 |
| n8n / Node-RED | 可视化 | 不适合写Python量化逻辑 |
| Hermes Agent | 刚好合适 | 相对新，文档在完善中 |
Hermes Agent打动我的核心设计：
1. 技能系统（Skills）
每项自动化能力封装为一个”技能”——本质上是一个Python包，有明确的输入输出、依赖声明和配置。这和cron脚本的本质区别在于：技能是可组合的、可复用的、可被AI调用的。
skills/
├── data-downloader/ # 数据下载技能
├── duckdb-sync/ # DuckDB同步技能
├── daily-report/ # 日报生成技能
├── market-monitor/ # 市场监控技能
├── ima-uploader/ # 飞书IMA上传技能
├── memory-updater/ # 记忆更新技能
└── ...2. Cron调度系统
Hermes Agent内置cron调度器，支持秒级精度（比系统cron的分钟级更细）。关键设计：
- 自动重试：失败自动重试3次，间隔递增（30s→2m→5m）
- 上下文窗口：每次执行都带前一次的结果上下文，类似Agent的”记忆”
- 飞书通知：执行结果自动推送到飞书（成功/失败/警告分级）
- 依赖链：任务A失败则跳过任务B（如：数据没下载完，就不跑日报）
3. 飞书推送集成
所有自动化任务的输出都通过飞书机器人推送到指定群聊：
📊 早间检查报告（06:00）
✅ 数据状态：正常（缺失0只）
✅ 磁盘使用：42%（安全）
✅ 网关连通：OK
⚠️ 信号：沪深300 10日涨幅+2.3%（超阈值）
📈 增量数据下载完成（18:02）
4589只股票更新完毕
28只可转债更新完毕
耗时：2分18秒这样我就不用每天登录服务器检查了——异常直接钉到我手机上。
系统架构全景
整个系统只有三台机器：一台WSL笔记本做开发，一台Linux服务器跑自动化，一台NAS存原始数据。
┌─────────────────────────────────────────────────┐
│ Linux Server │
│ │
│ Hermes Agent (主进程) │
│ ┌─────────────────────────────────┐ │
│ │ Cron Scheduler (50+条规则) │ │
│ │ ├─ 每日06:00 → morning-check │ │
│ │ ├─ 每日18:00 → data-download │ │
│ │ ├─ 每日 → bluechip-v2-signal │ │
│ │ ├─ 每周一 → weekly-report │ │
│ │ ├─ 周六 → convertible-report │ │
│ │ └─ 周日 → memory-update │ │
│ └─────────────────────────────────┘ │
│ │
│ Skills (Python + SQL) │
│ └─→ DuckDB (quant_v2.duckdb) │
│ └─→ CSV (增量存储) │
│ │
│ Outputs │
│ ├─→ 飞书机器人 (实时推送) │
│ ├─→ IMA知识库 (对话记录) │
│ └─→ GitHub (自动备份) │
└─────────────────────────────────────────────────┘数据流
pytdx（通达信接口）
↓ CSV文件（每只股票独立文件，ZSTD压缩）
↓ DuckDB（每日18:00同步，去重合并）
↓ SQL查询（策略信号、估值计算、日报生成）
↓ 飞书推送（结果格式化输出）技术栈
| 组件 | 技术 | 用途 |
|---|---|---|
| 自动化框架 | Hermes Agent | 技能管理、cron调度、上下文传递 |
| 数据库 | DuckDB | OLAP查询引擎，单文件294MB |
| 数据源 | pytdx | 通达信行情接口，免费 |
| 通知 | 飞书Webhook | 实时推送+分级告警 |
| 存储 | CSV (ZSTD) | 增量持久化，意外恢复 |
| 知识库 | IMA | 归档对话记录 |
核心自动化任务详解
目前系统运行着 20+ 条cron规则，覆盖量化投资的全部日常操作。以下逐一介绍核心任务。
每日06:00 — 早间检查
这是每天的第一道防线。
# morning_check 技能核心逻辑
def morning_check():
report = []
# 1. 数据完整性检查
missing_stocks = check_missing_data()
if missing_stocks:
report.append(f"⚠️ 数据缺失: {len(missing_stocks)}只股票")
else:
report.append("✅ 数据完整性: 正常")
# 2. 磁盘空间检查
disk_usage = get_disk_usage()
report.append(f"📀 磁盘使用: {disk_usage}%")
if disk_usage > 85:
report.append("🔴 磁盘告警！请清理")
# 3. 网关连通性
gateway_ok = ping_gateway()
report.append(f"🌐 网关: {'OK' if gateway_ok else 'FAILED'}")
# 4. 昨日成交量概览（异常检测）
volume_alert = check_volume_anomaly()
if volume_alert:
report.append(f"📊 异常放量: {volume_alert}")
# 5. 信号预警阈值检查
signals = check_signal_thresholds()
for s in signals:
if s['level'] == 'warning':
report.append(f"⚡ {s['name']}: {s['value']}")
return report运行结果示例（飞书推送）：
🌅 2026-06-17 早间检查报告
✅ 数据完整性：4589/4589 正常
✅ 磁盘使用：38%（安全）
✅ 网关连通：119.147.212.81 OK
✅ 昨日成交量：正常波动
✅ 无预警信号
⏰ 耗时：12秒为什么在06:00？ A股9:30开盘，6点检查有足够时间处理异常。如果发现问题，我可以7点起床处理，8点前搞定。
每日18:00 — 增量数据下载
收盘后（15:00）等数据稳定了再下载。
# data_downloader 技能核心
def incremental_download():
api = TDHQ()
api.connect('119.147.212.81', 7709)
total = len(stock_list)
success = 0
failed = []
for i, code in enumerate(stock_list):
try:
# 检查是否已有最新数据
latest = get_latest_date(code)
today = get_today_trading_date()
if latest >= today:
continue # 已是最新
# 下载增量
bars = api.get_security_bars(9, market_of(code), code, 0, 10)
if bars:
save_to_csv(code, bars)
success += 1
except Exception as e:
failed.append((code, str(e)))
# 每500只重连
if (i + 1) % 500 == 0:
api.disconnect()
api.connect('119.147.212.81', 7709)
# 同步到DuckDB
duckdb_sync()
return {
'total': total,
'success': success,
'failed': failed,
'duration': time_elapsed()
}优化点： 不是全量重下载，而是增量更新。只下载缺失的最多10条K线，4589只股票平均耗时2-3分钟。
每周一19:00 — 全市场估值周报
周一收盘后生成一份完整的估值周报：
-- DuckDB估值查询
SELECT
code,
name,
pe_ttm,
pb,
dividend_yield,
market_cap,
CASE
WHEN pe_ttm < 15 AND pb < 1.5 THEN '低估'
WHEN pe_ttm BETWEEN 15 AND 30 THEN '合理'
WHEN pe_ttm > 50 THEN '高估'
ELSE '关注'
END AS valuation_level
FROM stock_daily
JOIN fundamentals ON stock_daily.code = fundamentals.code
WHERE date = (SELECT MAX(date) FROM stock_daily)
ORDER BY market_cap DESC;输出包含：
- 全市场PE/PB分布直方图
- 各行业估值百分位
- 低估股票TOP20列表
- 与上月同期的估值变化对比
全部通过飞书富文本消息推送，一周一次，不看也行，但看了就有全局感。
每日 — 蓝筹v2双轨信号监控
这是一个实盘策略的信号系统，双轨并行：
轨1：趋势跟踪
当 MA20 > MA60 且 成交量 > MA5(成交量)*1.5 → 信号: 多
当 MA20 < MA60 且 成交量萎缩 → 信号: 观望轨2：均值回归
当 (收盘价 - MA20) / MA20 < -0.08 → 信号: 超卖（买入窗口）
当 (收盘价 - MA20) / MA20 > 0.08 → 信号: 超买（卖出窗口）每个交易日17:30运行，输出当前持仓股票的信号矩阵：
📡 蓝筹v2双轨信号 | 2026-06-17
贵州茅台(600519)
趋势跟踪: 🟢 多头（MA20>MA60）
均值回归: ⚪ 中性（偏离度+2.1%）
综合: ✅ 持有
招商银行(600036)
趋势跟踪: 🔴 空头（MA20<MA60）
均值回归: 🟢 超卖（偏离度-9.8%）
综合: ⚠️ 关注加仓点
中国平安(601318)
趋势跟踪: 🟢 多头
均值回归: 🟢 超卖
综合: ✅ 强烈看多每日 — 对话记录上传IMA知识库
Hermes Agent的一大特色：所有与Agent的对话历史，自动归档到知识库。
# ima_uploader 技能
def upload_conversations():
"""将当日量化相关对话上传到IMA知识库"""
conversations = get_today_conversations()
for conv in conversations:
if conv['category'] == 'quant':
# 提取关键决策点
key_decisions = extract_decisions(conv['content'])
# 格式化上传
ima_doc = {
'title': f"量化决策记录 - {conv['date']}",
'content': format_for_ima(conv['content']),
'tags': key_decisions,
'category': '量化研究'
}
upload_to_ima(ima_doc)
return len(conversations)半年下来，IMA知识库里积累了 600+ 篇量化相关的对话记录。需要回溯某个策略的决策过程时，直接搜索就行——比翻聊天记录快10倍。
周六 — 可转债推荐周报
每周末用DuckDB+SQL筛选可转债：
-- 可转债双低策略筛选
SELECT
bond_code,
bond_name,
stock_code,
(convert_value - current_price) / convert_value AS premium_rate,
current_price AS bond_price,
convert_value,
remaining_size,
-- 双低值 = 价格 + 溢价率*100
(current_price + premium_rate * 100) AS double_low_score
FROM convertible_bonds
WHERE
current_price < 130 -- 价格适中
AND premium_rate < 0.3 -- 溢价率<30%
AND remaining_size > 1 -- 剩余规模>1亿（流动性）
AND DATEDIFF('day', update_date, CURRENT_DATE) < 3
ORDER BY double_low_score ASC
LIMIT 20;输出格式：
📋 可转债双低周报 | 2026年第25周
排名 债券名称 现价 溢价率 双低值 剩余规模
1 本钢转债 108.5 2.1% 110.6 8.2亿
2 岩土转债 112.3 4.5% 116.8 4.5亿
3 大秦转债 115.8 3.2% 119.0 12.3亿
...
📊 本周建议关注：本钢转债、岩土转债、大秦转债
🔄 与上周对比：3只新入选，2只剔除周日 — 每周记忆+Wiki综合更新
这是最”AI Agent”的任务——Agent自己整理本周的量化研究进展，更新到知识库中。
# memory_updater 技能
def weekly_memory_update():
# 1. 汇总本周对话记录
weekly_convos = get_weekly_conversations()
# 2. 提取关键发现
findings = []
for convo in weekly_convos:
findings.extend(extract_key_findings(convo))
# 3. 更新Hermes Agent的记忆
for finding in findings:
add_to_memory(
key=finding['topic'],
value=finding['summary'],
category='quant_research',
ttl=timedelta(days=30) # 30天有效期
)
# 4. 更新Wiki知识库
weekly_summary = generate_weekly_summary(weekly_convos)
update_wiki(f"量化研究周报 {week_number}", weekly_summary)
return {
'findings': len(findings),
'memory_entries': len(weekly_convos),
'summary_length': len(weekly_summary)
}这解决了量化研究中的隐形知识流失问题——以前我和AI讨论完一个策略，下周就忘了细节。现在Agent会自己整理、归档、总结，下次讨论时自动引用历史上下文。
关键实现细节
技能系统的三层设计
每个Hermes Agent技能都按以下三层组织：
my-skill/
├── skill.json # 技能声明（名称、依赖、输入输出schema）
├── main.py # 核心逻辑
└── config.yaml # 运行时配置（token、路径、参数）skill.json 示例：
{
"name": "data-downloader",
"version": "2.0.0",
"description": "增量下载A股/可转债/ETF行情数据",
"dependencies": ["pytdx>=1.5", "duckdb>=0.9"],
"inputs": {
"mode": {"type": "string", "enum": ["incremental", "full"], "default": "incremental"},
"market": {"type": "string", "enum": ["stock", "bond", "etf", "all"], "default": "all"}
},
"outputs": {
"status": {"type": "object"}
},
"schedule": {
"default": "0 18 * * 1-5"
}
}Cron调度的进阶配置
Hermes Agent的cron语法和标准cron基本一致，但增加了几个实用扩展：
# ~/.hermes/cron.yaml
tasks:
# 标准定时任务
- name: morning-check
schedule: "0 6 * * 1-5" # 工作日06:00
skill: morning-check
notify_on: [failure, warning] # 失败或警告才通知
- name: data-download
schedule: "0 18 * * 1-5" # 工作日18:00
skill: data-downloader
notify_on: [failure] # 只有失败才通知
- name: bluechip-signal
schedule: "30 15 * * 1-5" # 工作日15:30
skill: market-signal
args:
strategy: bluechip-v2
notify_on: [always] # 每次都推送
# 依赖链
- name: daily-report
schedule: "0 19 * * 1-5"
skill: daily-report
depends_on:
- data-download # 数据下载完成后才跑
retry: 3 # 失败重试3次
retry_interval: [30, 120, 300] # 30s→2m→5m
# 周报
- name: weekly-valuation
schedule: "0 19 * * 1" # 每周一19:00
skill: weekly-report
notify_on: [always]
- name: convertible-report
schedule: "0 10 * * 6" # 周六10:00
skill: convertible-report
notify_on: [always]
- name: memory-update
schedule: "0 9 * * 0" # 周日09:00
skill: memory-updater
notify_on: [always]飞书推送集成
推送通过飞书自定义机器人Webhook实现：
# feishu_notifier.py
import requests
import json
class FeishuNotifier:
def __init__(self, webhook_url):
self.webhook_url = webhook_url
def send_text(self, text, level="info"):
"""发送纯文本消息"""
tag = {"info": "💬", "warning": "⚠️", "error": "🔴"}
payload = {
"msg_type": "text",
"content": {
"text": f"{tag[level]} {text}"
}
}
return requests.post(self.webhook_url, json=payload)
def send_rich_text(self, title, blocks, level="info"):
"""发送富文本消息"""
content = []
for block in blocks:
if isinstance(block, str):
content.append([{"tag": "text", "text": block}])
elif isinstance(block, dict):
content.append([block])
payload = {
"msg_type": "post",
"content": {
"post": {
"zh_cn": {
"title": title,
"content": content
}
}
}
}
return requests.post(self.webhook_url, json=payload)效果评估
系统上线运行 6个月 以来，自动化覆盖了量化投资的全部日常操作。
时间节省
| 任务 | 手动耗时 | 自动化后 | 每周节省 |
|---|---|---|---|
| 数据下载 | 10min/天 | 3min/天（后台） | ~35min |
| 数据完整性检查 | 5min/天 | 0 | ~25min |
| 日报生成 | 15min/天 | 0 | ~75min |
| 周报生成 | 60min/周 | 0 | ~60min |
| 信号监控 | 10min/天 | 0 | ~50min |
| 知识库归档 | 20min/周 | 0 | ~20min |
| 系统维护检查 | 10min/天 | 0 | ~50min |
| 总计 | ~5h/周 | ~15min/周 | ~4.75h/周 |
每周省出近5小时——这些时间全部回到了策略研究和阅读上。
自动化覆盖率
✅ 数据采集 100% — 每日自动下载，7×24无人值守
✅ 数据清洗 100% — 去重、前复权、异常值处理全自动
✅ 日报报告 95% — 模板化报告自动生成，特殊行情手动补充
✅ 信号监控 100% — 双轨信号自动计算+预警推送
✅ 知识管理 90% — 对话自动归档，关键决策点自动提取
⚠️ 策略回测 60% — 常规回测自动化，新策略开发仍需人工可靠性数据
- 任务总数：6个月 × 22天/月 × 15个工作日任务 = ~1,980次
- 成功率：97.3%（故障主要来自网络波动）
- 自动恢复成功：故障中82%自动重试恢复
- 需要人工介入：18次，平均每月3次（主要是pytdx主站维护）
踩坑记录
任何自动化系统都是在”坑”上建起来的。以下是最痛的几个。
🕳️ 坑1：飞书Webhook Token权限
问题：飞书自定义机器人的Webhook URL包含token，但这个token有有效期限制。某天突然推送全部失败，排查半天发现token过期了。
解决：
- 不再用”自定义机器人”，改用飞书应用 + 机器人API，通过OAuth 2.0获取token
- 添加token自动刷新逻辑（有效期2小时，提前30分钟刷新）
- 推送失败时自动重试获取新token
def get_feishu_token(app_id, app_secret):
"""获取飞书tenant_access_token"""
url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
resp = requests.post(url, json={
"app_id": app_id,
"app_secret": app_secret
})
return resp.json().get("tenant_access_token")🕳️ 坑2：代理不稳定导致下载中断
问题：WSL的代理配置在每次重启后会丢失，导致数据下载任务连接通达信超时。最惨的一次：代理挂了一天，结果连续3天数据没下载，最后手动补了2小时。
解决：
- 在Hermes Agent技能中硬编码备用代理列表，主代理失效自动切换
- 添加连接超时+重试机制（每次请求超时5秒，重试3次）
- 早间检查增加代理连通性测试，有问题直接推送告警
🕳️ 坑3：pytdx限流
问题：单IP对通达信主站的并发请求有限制。之前没控制并发，高峰期被临时封了IP 30分钟。
解决：
- 每500只股票断开重连（释放连接数）
- 每次请求间隔0.1秒
- 被封IP时自动切换到备用主站IP
🕳️ 坑4：DuckDB写入冲突
问题：cron任务和手动查询同时写入DuckDB时，偶尔出现database is locked错误。
解决：
- 所有写操作使用DuckDB的WAL模式
- 写操作超时设置30秒
- 任务之间通过Hermes cron的依赖链确保不会并发写入
-- DuckDB配置优化
SET enable_progress_bar = false;
SET preserve_insertion_order = false;
SET max_expression_depth = 1000;
SET threads = 8;
SET memory_limit = '4GB';🕳️ 坑5：时间同步问题
问题：WSL的时间与宿主机Windows偶尔不同步，导致cron任务的datetime.now()比实际时间晚了几分钟。数据下载任务因此错过了收盘后的最佳下载窗口。
解决：
- 在早间检查脚本中添加
ntpdate同步 - 所有时间相关逻辑改为使用网络时间（通过HTTP响应头获取）
- cron规则中的时间全部用UTC+8写死
未来扩展方向
当前的自动化系统已经稳定运行，但还有很多可以提升的方向：
1. 策略回测自动化（下一步）
目前策略回测还需要手动触发。计划用Hermes Agent的runbook功能，把回测流程封装为可调用的技能：
# 策略A：布林带均值回归
runbook = Runbook([
Step("拉取最新数据", data_downloader),
Step("计算布林带信号", bollinger_signal),
Step("模拟交易", backtest_engine, params={"strategy": "bollinger"}),
Step("生成回测报告", report_generator),
Step("推送飞书", feishu_notifier)
])
weekly_schedule = "0 20 * * 5" # 周五晚8点2. 多数据库联邦查询
当前只有DuckDB一个数据源。计划接入更多：
- PostgreSQL：存放交易记录和账户数据
- SQLite：存放策略参数和运行日志
- CSV/Parquet：原始数据快照
Hermes Agent通过DuckDB的ATTACH和mysql_scanner/postgres_scanner实现联邦查询。
3. AI驱动的异常检测
目前异常检测靠硬编码阈值（如：磁盘>85%告警）。计划用Hermes Agent的LLM能力做更智能的异常判断：
- 数据异常：突然某行业全部缺失 → 可能是数据提供商问题
- 信号异常：所有策略同时发出买入信号 → 可能是市场异常波动
- 日志异常：错误日志突然增多 → 可能是系统问题
4. 自动调参优化
利用Hermes Agent的Memory功能，记录策略在不同参数下的表现，自动搜索最优参数组合：
# 自动调参任务（计划中）
- name: auto-optimize
schedule: "0 3 * * 0" # 周日凌晨3点
skill: strategy-optimizer
args:
strategy: bluechip-v2
search_space:
ma_period: [10, 20, 30, 60]
threshold: [5, 8, 10, 15]
metric: sharpe_ratio
method: grid_search5. 灾难恢复自动化
当前如果服务器挂了，需要手动重建全部环境。计划把整个恢复流程也做成Hermes技能：
灾难恢复技能（计划中）
├─ 检查备份状态（GitHub + NAS双备份）
├─ 恢复 DuckDB 数据库
├─ 恢复 CSV 数据文件
├─ 重建 Hermes Agent 环境
├─ 恢复 cron 配置
└─ 验证数据完整性总结
这套基于Hermes Agent的量化自动化系统，让我从每天2小时的”搬砖”中解脱出来。关键收获：
- AI Agent≠大语言模型——Hermes Agent解决的是”自动化编排”问题，LLM只是其中一环
- 技能可组合设计让自动化从”脚本集合”升级为”能力系统”
- 飞书推送改变了工作流——从”主动检查”变成”异常驱动”
- cron调度的进阶用法——依赖链、自动重试、上下文传递，这些原生cron做不到
如果你是个人量化投资者，还在手动下载数据、手动生成报告、每天登录服务器检查——强烈推荐试试Hermes Agent。它不是为了企业级ETL设计的，但正好适合一个人+一台服务器+一堆策略的量化场景。
📚 相关文章推荐
- Hermes Agent 优化迭代实录：一个AI Agent的持续进化之路 — 同为Hermes Agent实战系列，记录23次迭代优化的完整过程
- DuckDB搭建A股量化数据库：4589只股票本地数据库实战教程 — Agent每日调度的数据底座，自动化任务的核心数据源
- 个人博客从零到上线全记录：Astro + GitHub + Cloudflare + 自定义域名 — Agent实战系列延伸，量化博客从零到上线的搭建记录
风险提示：本文所有内容均为技术分享，不构成投资建议。历史回测不代表未来收益。文中提到的策略仅为示例，不构成任何投资建议。