OpenClaw Trading - 生存压力驱动的量化交易系统设计文档
概述
结合 ClawWork 的生存压力机制 + TradingAgents 的多智能体架构 + abu 的因子系统，创建一个必须为自己的决策付费的交易 Agent 系统。
1. 系统架构
┌─────────────────────────────────────────────────────────────────┐
│ OpenClaw Trading │
│ 生存压力驱动的量化系统 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 资金层 │◄──►│ Agent 层 │◄──►│ 市场层 │ │
│ │ Capital │ │ Multi-Agent │ │ Market │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 生存压力引擎 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │ │
│ │ │ 成本计算 │ │ 收益评估 │ │ 生存状态管理 │ │ │
│ │ │ Cost │ │ Reward │ │ Life State │ │ │
│ │ └──────────┘ └──────────┘ └──────────────────┘ │ │
│ └──────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
2. 核心机制设计
2.1 经济压力模型（借鉴 ClawWork）
class TradingEconomicTracker:
"""
交易Agent经济追踪器
每个Agent必须为自己的决策付费
"""
def __init__(self,
agent_id: str,
initial_capital: float = 10000.0, # 启动资金 $10,000
token_cost_per_1m_input: float = 2.5,
token_cost_per_1m_output: float = 10.0,
trade_fee_rate: float = 0.001): # 交易手续费 0.1%
self.balance = initial_capital
self.token_costs = 0.0
self.trade_costs = 0.0
self.realized_pnl = 0.0
# 生存状态阈值
self.thresholds = {
'thriving': initial_capital * 1.5, # 盈利 50%+
'stable': initial_capital * 1.1, # 盈利 10%+
'struggling': initial_capital * 0.8, # 亏损 20%+
'bankrupt': initial_capital * 0.3 # 亏损 70%+
}
def calculate_decision_cost(self,
tokens_input: int,
tokens_output: int,
market_data_calls: int) -> float:
"""
计算每次决策的成本
"""
llm_cost = (tokens_input / 1e6 * self.token_cost_per_1m_input +
tokens_output / 1e6 * self.token_cost_per_1m_output)
data_cost = market_data_calls * 0.01 # 每次数据调用 $0.01
total_cost = llm_cost + data_cost
self.token_costs += total_cost
self.balance -= total_cost
return total_cost
def calculate_trade_cost(self,
trade_value: float,
is_win: bool,
win_amount: float = 0.0,
loss_amount: float = 0.0) -> dict:
"""
计算交易成本和收益
"""
fee = trade_value * self.trade_fee_rate
self.trade_costs += fee
self.balance -= fee
pnl = win_amount - loss_amount - fee
self.realized_pnl += pnl
self.balance += pnl
return {
'fee': fee,
'pnl': pnl,
'balance': self.balance,
'status': self.get_survival_status()
}
def get_survival_status(self) -> str:
"""获取生存状态"""
if self.balance >= self.thresholds['thriving']:
return '🚀 thriving' # 繁荣 - 可扩张交易规模
elif self.balance >= self.thresholds['stable']:
return '💪 stable' # 稳定 - 正常交易
elif self.balance >= self.thresholds['struggling']:
return '⚠️ struggling' # 挣扎 - 只能做小单
elif self.balance >= self.thresholds['bankrupt']:
return '🔴 critical' # 危急 - 只能模拟交易
else:
return '💀 bankrupt' # 破产 - 被淘汰
2.2 多 Agent 角色设计（借鉴 TradingAgents）
class TradingAgentTeam:
"""
交易Agent团队
每个Agent都有自己的资金账户和生存压力
"""
def __init__(self):
self.agents = {
# 分析师团队 - 成本低，收费少
'market_analyst': AnalystAgent(
role='market',
decision_cost=0.05, # 每次分析 $0.05
min_balance=50
),
'sentiment_analyst': AnalystAgent(
role='sentiment',
decision_cost=0.08,
min_balance=80
),
'fundamental_analyst': AnalystAgent(
role='fundamental',
decision_cost=0.10,
min_balance=100
),
# 研究员团队 - 中等成本
'bull_researcher': ResearcherAgent(
stance='bull',
decision_cost=0.15,
min_balance=150
),
'bear_researcher': ResearcherAgent(
stance='bear',
decision_cost=0.15,
min_balance=150
),
# 风险管理 - 高成本但必要
'risk_manager': RiskManagerAgent(
decision_cost=0.20,
min_balance=200
),
# 交易员 - 执行决策，成本最高
'trader': TraderAgent(
decision_cost=0.30,
min_balance=500,
trade_fee_rate=0.001
)
}
2.3 工作-交易权衡机制（ClawWork 核心机制）
class WorkTradeBalance:
"""
工作-交易权衡系统
Agent需要决定：立即交易赚钱 vs 学习提升能力
"""
def decide_activity(self, agent: Agent) -> str:
"""
根据当前经济状态决定是交易还是学习
"""
status = agent.economic_tracker.get_survival_status()
skill_level = agent.skill_level
win_rate = agent.historical_win_rate
# 决策逻辑
if status == '💀 bankrupt':
return 'liquidate' # 清仓，停止交易
elif status == '🔴 critical':
# 危急状态：只能做最有把握的交易
if win_rate > 0.7 and skill_level > 0.8:
return 'conservative_trade'
else:
return 'paper_trade' # 模拟交易，学习为主
elif status == '⚠️ struggling':
# 挣扎状态：谨慎交易，适当学习
if win_rate < 0.5:
return 'learn' # 胜率低，先学习
else:
return 'selective_trade' # 选择性交易
elif status == '💪 stable':
# 稳定状态：正常交易 + 适度学习
if skill_level < 0.6:
return 'learn' # 投资自己
else:
return 'normal_trade'
elif status == '🚀 thriving':
# 繁荣状态：可以承担更多风险
if skill_level < 0.9:
return 'aggressive_learn' # 大量投资学习
else:
return 'aggressive_trade' # 扩大交易规模
2.4 学习投资系统
class LearningInvestment:
"""
学习投资系统
Agent可以投资学习来提升交易能力
"""
LEARNING_COURSES = {
'technical_analysis': {
'cost': 100.0, # 学费 $100
'duration_days': 7, # 学习周期 7天
'skill_improvement': 0.1, # 技能提升 10%
'win_rate_boost': 0.05 # 胜率提升 5%
},
'risk_management': {
'cost': 150.0,
'duration_days': 10,
'skill_improvement': 0.15,
'max_drawdown_reduction': 0.1
},
'market_psychology': {
'cost': 200.0,
'duration_days': 14,
'skill_improvement': 0.2,
'sentiment_accuracy_boost': 0.1
},
'advanced_strategies': {
'cost': 500.0,
'duration_days': 30,
'skill_improvement': 0.3,
'new_strategy_unlock': True
}
}
def enroll_course(self, agent: Agent, course_name: str) -> bool:
"""
报名学习课程
"""
course = self.LEARNING_COURSES[course_name]
# 检查是否有足够资金
if agent.balance < course['cost'] * 1.5: # 保留50%安全边际
return False
# 扣除学费
agent.balance -= course['cost']
# 开始学习
agent.learning_status = {
'course': course_name,
'start_date': datetime.now(),
'end_date': datetime.now() + timedelta(days=course['duration_days']),
'expected_improvement': course['skill_improvement']
}
return True
2.5 交易因子插件系统（借鉴 abu）
class FactorPluginSystem:
"""
因子插件系统
Agent可以购买/解锁因子来提升交易能力
"""
AVAILABLE_FACTORS = {
# 基础因子 - 免费
'moving_average_cross': {
'cost': 0,
'type': 'buy',
'description': '均线金叉策略'
},
'rsi_oversold': {
'cost': 0,
'type': 'buy',
'description': 'RSI超卖反弹'
},
# 进阶因子 - 付费解锁
'bollinger_squeeze': {
'cost': 50.0,
'type': 'buy',
'description': '布林带挤压突破'
},
'macd_divergence': {
'cost': 80.0,
'type': 'buy',
'description': 'MACD背离信号'
},
# 高级因子 - 昂贵但强大
'machine_learning_pred': {
'cost': 500.0,
'type': 'buy',
'description': '机器学习预测模型'
},
'sentiment_momentum': {
'cost': 300.0,
'type': 'buy',
'description': '情绪动量策略'
},
# 卖出因子
'atr_trailing_stop': {
'cost': 100.0,
'type': 'sell',
'description': 'ATR追踪止损'
},
'time_decay_exit': {
'cost': 150.0,
'type': 'sell',
'description': '时间衰减退出'
}
}
def purchase_factor(self, agent: Agent, factor_name: str) -> bool:
"""
购买因子
"""
factor = self.AVAILABLE_FACTORS[factor_name]
if agent.balance < factor['cost'] * 1.2: # 保留20%缓冲
return False
agent.balance -= factor['cost']
agent.unlocked_factors.append(factor_name)
return True
2.6 风险评估与拦截系统（借鉴 abu UMP）
class SurvivalRiskManager:
"""
生存风险管理系统
防止Agent因高风险交易而破产
"""
def evaluate_trade_risk(self, agent: Agent, trade_signal: dict) -> dict:
"""
评估交易对Agent生存的风险
"""
balance = agent.balance
status = agent.economic_tracker.get_survival_status()
# 风险指标
position_size = trade_signal['position_size']
stop_loss = trade_signal['stop_loss']
max_loss = position_size * stop_loss
risk_assessment = {
'approved': False,
'risk_level': 'unknown',
'max_position_size': 0,
'reason': ''
}
# 根据生存状态限制风险
if status == '💀 bankrupt':
risk_assessment['reason'] = 'Agent已破产，禁止交易'
return risk_assessment
elif status == '🔴 critical':
# 危急状态：最大损失不能超过余额的1%
max_risk = balance * 0.01
if max_loss > max_risk:
risk_assessment['reason'] = f'风险过高，最大允许损失: ${max_risk:.2f}'
risk_assessment['max_position_size'] = max_risk / stop_loss
else:
risk_assessment['approved'] = True
risk_assessment['risk_level'] = 'extreme_low'
elif status == '⚠️ struggling':
max_risk = balance * 0.03 # 3%风险
if max_loss > max_risk:
risk_assessment['max_position_size'] = max_risk / stop_loss
else:
risk_assessment['approved'] = True
risk_assessment['risk_level'] = 'low'
elif status == '💪 stable':
max_risk = balance * 0.05 # 5%风险
if max_loss > max_risk:
risk_assessment['max_position_size'] = max_risk / stop_loss
else:
risk_assessment['approved'] = True
risk_assessment['risk_level'] = 'medium'
elif status == '🚀 thriving':
max_risk = balance * 0.10 # 10%风险
if max_loss > max_risk:
risk_assessment['max_position_size'] = max_risk / stop_loss
else:
risk_assessment['approved'] = True
risk_assessment['risk_level'] = 'high'
return risk_assessment
3. 核心工作流程
class OpenClawTradingWorkflow:
"""
OpenClaw Trading 核心工作流
"""
async def execute_trading_cycle(self, agent_team: TradingAgentTeam, symbol: str):
"""
执行一个交易周期
"""
# Step 1: 检查每个Agent的生存状态
for agent_name, agent in agent_team.agents.items():
status = agent.economic_tracker.get_survival_status()
if status == '💀 bankrupt':
logger.warning(f'{agent_name} 已破产，跳过该Agent')
continue
# Step 2: 决定活动（交易还是学习）
work_trade = WorkTradeBalance()
activities = {}
for agent_name, agent in agent_team.agents.items():
activities[agent_name] = work_trade.decide_activity(agent)
# Step 3: 如果决定学习，执行学习
for agent_name, activity in activities.items():
if 'learn' in activity:
await self.execute_learning(agent_team.agents[agent_name])
# Step 4: 市场分析（分析师团队）
analysis_results = {}
for analyst_name in ['market_analyst', 'sentiment_analyst', 'fundamental_analyst']:
analyst = agent_team.agents[analyst_name]
if activities[analyst_name] in ['normal_trade', 'selective_trade', 'conservative_trade']:
# 扣除分析成本
cost = analyst.economic_tracker.calculate_decision_cost(
tokens_input=2000,
tokens_output=500,
market_data_calls=5
)
analysis_results[analyst_name] = await analyst.analyze(symbol)
logger.info(f'{analyst_name} 分析完成，成本: ${cost:.4f}')
# Step 5: 研究员辩论（看多 vs 看空）
if activities['bull_researcher'] in ['normal_trade', 'selective_trade']:
bull_view = await agent_team.agents['bull_researcher'].debate(
analysis_results, 'bull'
)
if activities['bear_researcher'] in ['normal_trade', 'selective_trade']:
bear_view = await agent_team.agents['bear_researcher'].debate(
analysis_results, 'bear'
)
# Step 6: 风险评估
risk_assessment = await agent_team.agents['risk_manager'].assess_risk(
bull_view, bear_view, symbol
)
# Step 7: 交易决策
if risk_assessment['approved']:
trader = agent_team.agents['trader']
# 生存风险管理
survival_risk = SurvivalRiskManager()
trade_approval = survival_risk.evaluate_trade_risk(
trader, risk_assessment['trade_signal']
)
if trade_approval['approved']:
# 执行交易
result = await trader.execute_trade(
symbol,
risk_assessment['trade_signal'],
max_position=trade_approval.get('max_position_size')
)
# 计算交易后的经济状态
trade_cost_result = trader.economic_tracker.calculate_trade_cost(
trade_value=result['value'],
is_win=result['pnl'] > 0,
win_amount=max(0, result['pnl']),
loss_amount=max(0, -result['pnl'])
)
logger.info(f'交易执行: {result}')
logger.info(f'经济状况: {trade_cost_result}')
# Step 8: 反思与学习（BM25 记忆系统）
await agent_team.reflect_and_learn()
return {
'activities': activities,
'analysis': analysis_results,
'trades': result if 'result' in locals() else None,
'economic_status': {
name: agent.economic_tracker.get_survival_status()
for name, agent in agent_team.agents.items()
}
}
4. 可视化仪表板
class SurvivalDashboard:
"""
生存状态实时仪表板
"""
def render(self, agent_team: TradingAgentTeam):
"""
渲染实时仪表板
"""
console = Console()
# 创建Agent状态表格
table = Table(title="Agent 生存状态")
table.add_column("Agent", style="cyan")
table.add_column("余额", justify="right")
table.add_column("状态", justify="center")
table.add_column("胜率", justify="right")
table.add_column("技能等级", justify="right")
table.add_column("已解锁因子", justify="center")
for name, agent in agent_team.agents.items():
status = agent.economic_tracker.get_survival_status()
status_color = {
'🚀 thriving': 'green',
'💪 stable': 'blue',
'⚠️ struggling': 'yellow',
'🔴 critical': 'red',
'💀 bankrupt': 'dim'
}.get(status, 'white')
table.add_row(
name,
f"${agent.balance:,.2f}",
f"[{status_color}]{status}[/{status_color}]",
f"{agent.win_rate:.1%}",
f"{agent.skill_level:.1%}",
str(len(agent.unlocked_factors))
)
console.print(table)
# 经济压力曲线图
self.render_pressure_chart(agent_team)
# 交易盈亏分布
self.render_pnl_distribution(agent_team)
5. 配置示例
{
"openclaw_trading": {
"initial_capital": {
"market_analyst": 1000,
"sentiment_analyst": 1000,
"fundamental_analyst": 1000,
"bull_researcher": 2000,
"bear_researcher": 2000,
"risk_manager": 3000,
"trader": 10000
},
"cost_structure": {
"llm_input_per_1m": 2.5,
"llm_output_per_1m": 10.0,
"market_data_per_call": 0.01,
"trade_fee_rate": 0.001
},
"survival_thresholds": {
"thriving_multiplier": 1.5,
"stable_multiplier": 1.1,
"struggling_multiplier": 0.8,
"bankrupt_multiplier": 0.3
},
"learning_courses": {
"enabled": true,
"auto_enroll": false,
"min_balance_ratio": 1.2
},
"factor_market": {
"enabled": true,
"free_factors": ["moving_average_cross", "rsi_oversold"],
"premium_factors": ["machine_learning_pred", "sentiment_momentum"]
}
}
}
6. 系统特点总结
| 特性 | 设计来源 | 说明 |
|---|---|---|
| 💰 经济压力 | ClawWork | 每个Agent必须为自己的token和交易付费 |
| 🎯 生存状态 | ClawWork | 5级生存状态，影响交易权限和风险承受 |
| 📚 学习投资 | ClawWork | Agent可投资学习提升能力，但需权衡成本 |
| 🤖 多Agent协作 | TradingAgents | 分析师、研究员、风险管理、交易员分工 |
| 🧠 记忆系统 | TradingAgents | BM25离线记忆，从过往交易中学习 |
| 📊 因子插件 | abu | 可购买解锁的交易策略因子 |
| 🛡️ 风险拦截 | abu UMP | 基于生存状态的动态风险限制 |
7. 实现路线图
第一阶段：基础框架
- 实现基础经济追踪器
- 单Agent交易能力
- 生存状态管理
- 基础CLI界面
第二阶段：多Agent协作
- 添加多Agent协作架构
- 实现辩论机制
- 工作流编排
- 记忆系统集成
第三阶段：高级功能
- 因子市场系统
- 学习投资系统
- 风险管理拦截
- 可视化仪表板
第四阶段：生产就绪
- 完善监控系统
- 添加回测能力
- 性能优化
- 文档和示例
8. 参考项目
| 项目 | 核心借鉴 | 路径 |
|---|---|---|
| ClawWork | 经济压力机制、生存状态 | /Users/cillin/workspeace/stock/reference/ClawWork |
| TradingAgents | 多智能体架构、BM25记忆 | /Users/cillin/workspeace/stock/reference/TradingAgents |
| abu | 因子插件系统、UMP风险拦截 | /Users/cillin/workspeace/stock/reference/abu |
| daily_stock_analysis | 数据源管理、通知推送 | /Users/cillin/workspeace/stock/reference/daily_stock_analysis |
| Lean | 回测引擎、性能优化 | /Users/cillin/workspeace/stock/reference/Lean |
9. 调研报告目录
所有参考项目的详细调研报告保存在：
/Users/cillin/workspeace/stock/report/
├── abu_report.md # 阿布量化系统
├── ClawWork_report.md # AI经济生存基准测试
├── daily_stock_analysis_report.md # 每日股票分析系统
├── Lean_report.md # 量化交易平台（待生成）
└── TradingAgents_report.md # 多智能体交易框架
10. Phase 4 实现完成
完成时间: 2026-02-25
已完成模块
| 任务 | 模块 | 文件数 | 状态 |
|---|---|---|---|
| TASK-045 | 策略框架基类 | 7 | ✅ 完成 |
| TASK-046 | 策略组合管理器 | 5 | ✅ 完成 |
| TASK-047 | 策略回测对比 | 5 | ✅ 完成 |
| TASK-048 | Agent学习记忆 | 3 | ✅ 完成 |
| TASK-049 | 策略优化器 | 6 | ✅ 完成 |
| TASK-050 | 进化算法集成 | 6 | ✅ 完成 |
代码统计
- Python 文件: 64 个
- 测试文件: 17 个
- 测试用例: 300+
功能验证
运行 python demo_phase4.py 验证所有功能:
- ✅ 策略框架基类 (Strategy, Signal, StrategyContext)
- ✅ 策略组合管理 (StrategyPortfolio, 权重分配)
- ✅ 策略回测对比 (ComparisonMetrics, StrategyComparator)
- ✅ Agent学习记忆 (LearningMemory, BM25Index)
- ✅ 策略优化器 (GridSearch, RandomSearch, Bayesian)
- ✅ 进化算法 (GeneticAlgorithm, EvolutionEngine, NSGA2)
实现路线图更新
✅ 第一阶段：基础框架
- 实现基础经济追踪器
- 单Agent交易能力
- 生存状态管理
- 基础CLI界面
✅ 第二阶段：多Agent协作
- 添加多Agent协作架构
- 实现辩论机制
- 工作流编排
- 记忆系统集成
🔄 第三阶段：高级功能 (进行中)
- 因子市场系统
- 学习投资系统
- 风险管理拦截
- 可视化仪表板
⏳ 第四阶段：生产就绪
- 完善监控系统
- 添加回测能力
- 性能优化
- 文档和示例
设计文档版本: 1.1 Phase 4 完成时间: 2026-02-25 设计来源: Claude Code 基于多项目分析