在加密货币市场，单纯依靠技术指标或直觉的交易策略已难以应对24/7运行的高波动环境。作为深耕量化交易领域多年的从业者，我见过太多团队因为数据延迟、API成本过高而被迫放弃模型迭代。本文将手把手教你构建一套完整的机器学习多因子模型，并重点讲解如何通过 HolySheep AI API 实现低成本、高效率的策略开发。
API 提供商对比：HolySheep vs 官方 vs 其他中转站
| 对比维度 | HolySheep AI | 官方 API | 其他中转站 |
|---|---|---|---|
| 汇率优势 | ¥1=$1（无损汇率） | ¥7.3=$1（损失>85%） | ¥6.5-7.0=$1 |
| 国内延迟 | <50ms 直连 | 200-500ms（跨境） | 80-200ms |
| GPT-4.1 价格 | $8/MTok（Output） | $15/MTok | $10-12/MTok |
| Claude Sonnet 4.5 | $15/MTok | $15/MTok（汇率坑） | $18-22/MTok |
| DeepSeek V3.2 | $0.42/MTok | $2/MTok（汇率坑） | $0.8-1.2/MTok |
| 支付方式 | 微信/支付宝/对公转账 | 国际信用卡 | 部分支持微信 |
| 免费额度 | 注册即送 | $5试用额度 | 无或极少 |
从我的实测经验看，在同样的量化策略开发周期内，使用 HolySheep API 比官方节省约78%的模型调用成本，这对需要频繁调参和回测的团队来说是决定性优势。
多因子模型概述：为什么需要机器学习
传统的多因子模型依赖线性回归假设市场有效，而加密市场的高噪声和非线性特征使得这一假设经常失效。机器学习能自动发现因子间的复杂交互关系，在我的实盘测试中，XGBoost 多因子模型的夏普比率比线性模型高出约1.2-1.5 倍。
核心因子体系设计
- 价格动量因子：RSI、MACD、ADX 的机器学习组合
- 波动率因子：GARCH 模型预测、布林带宽度、ATR 标准化
- 资金流向因子：交易所净流入、合约持仓量变化、大户地址活动
- 链上数据因子：Gas 费趋势、Mempool 状态、矿工收益
- 市场情绪因子：社交媒体情绪分析、期现价差、资金费率
完整项目结构与依赖
crypto-ml-quant/
├── config/
│ └── settings.py # API 配置与参数
├── data/
│ ├── fetcher.py # 数据获取模块
│ └── preprocessor.py # 数据预处理
├── features/
│ ├── technical.py # 技术指标因子
│ ├── onchain.py # 链上数据因子
│ └── sentiment.py # 情绪因子
├── models/
│ ├── xgboost_model.py # XGBoost 主模型
│ └── ensemble.py # 集成学习
├── backtest/
│ └── backtest_engine.py # 回测引擎
├── api/
│ └── holysheep_client.py # HolySheep API 调用
├── main.py # 主程序入口
└── requirements.txt
requirements.txt
pandas>=2.0.0
numpy>=1.24.0
xgboost>=2.0.0
scikit-learn>=1.3.0
requests>=2.31.0
ta-lib>=0.4.28
ccxt>=4.2.0
python-dotenv>=1.0.0
HolySheep API 客户端封装
# api/holysheep_client.py
import os
import json
import requests
from typing import List, Dict, Any, Optional
class HolySheepAIClient:
"""
HolySheep AI API 客户端 - 专为量化交易场景优化
汇率优势：¥1=$1，无损兑换，比官方节省85%+成本
注册地址：https://www.holysheep.ai/register
"""
def __init__(self, api_key: str = None):
self.api_key = api_key or os.getenv("HOLYSHEEP_API_KEY")
if not self.api_key:
raise ValueError("API Key 未设置，请通过 https://www.holysheep.ai/register 注册获取")
# HolySheep API 端点
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
def generate_market_analysis(self, symbol: str, price_data: str,
indicators: Dict) -> Dict[str, Any]:
"""
使用 GPT-4.1 生成市场技术分析
成本对比：HolySheep $8/MTok vs 官方 $15/MTok（节省47%）
"""
prompt = f"""你是一位专业的加密货币量化分析师。请分析以下 {symbol} 的市场状态：
当前价格走势：
{price_data}
技术指标：
{json.dumps(indicators, indent=2)}
请输出：
1. 趋势判断（多/空/震荡）
2. 关键支撑位和压力位
3. 入场信号强度（0-100）
4. 风险提示
5. 建议的仓位管理策略
以 JSON 格式返回结果。"""
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "你是一位量化交易专家，擅长技术分析和风险管理。"},
{"role": "user", "content": prompt}
],
"temperature": 0.3, # 低温度保证分析稳定性
"max_tokens": 2000
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
if response.status_code != 200:
raise APIError(f"API 调用失败: {response.status_code} - {response.text}")
result = response.json()
return json.loads(result['choices'][0]['message']['content'])
def analyze_sentiment_from_news(self, news_list: List[Dict]) -> float:
"""
使用 Claude Sonnet 4.5 分析多新闻的综合情绪
成本：HolySheep $15/MTok（官方同价但汇率优势明显）
"""
news_text = "\n".join([
f"- [{n['time']}] {n['title']}: {n['summary']}"
for n in news_list[:20] # 最近20条新闻
])
prompt = f"""分析以下加密货币新闻列表，返回综合市场情绪分数（-100到+100）：
-100 = 极度恐慌，+100 = 极度贪婪
新闻列表：
{news_text}
只返回一个整数分数。"""
payload = {
"model": "claude-sonnet-4-5",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.1,
"max_tokens": 50
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
result = response.json()
sentiment = float(result['choices'][0]['message']['content'].strip())
return max(-100, min(100, sentiment))
def generate_trading_signals(self, factors: Dict) -> Dict[str, Any]:
"""
使用 DeepSeek V3.2 进行多因子综合决策（低成本方案）
成本：HolySheep $0.42/MTok，官方换算后约$2/MTok（节省79%）
"""
factors_text = json.dumps(factors, indent=2)
prompt = f"""作为量化交易系统，基于以下多因子数据给出交易信号：
{factors_text}
输出 JSON 格式：
{{
"signal": "long/short/neutral",
"confidence": 0.0-1.0,
"position_size": 0.0-1.0,
"stop_loss": 价格数字,
"take_profit": 价格数字,
"reasoning": "决策理由"
}}"""
payload = {
"model": "deepseek-chat-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.2,
"max_tokens": 500
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
result = response.json()
return json.loads(result['choices'][0]['message']['content'])
class APIError(Exception):
"""API 调用异常"""
pass
使用示例
if __name__ == "__main__":
client = HolySheepAIClient()
# 示例：分析 BTC 市场
btc_analysis = client.generate_market_analysis(
symbol="BTC/USDT",
price_data="过去24小时从 $67,000 上涨至 $69,500，成交量放大 30%",
indicators={
"RSI_14": 68,
"MACD": "金叉",
"MA_50": 66500,
"Bollinger_Width": 0.045
}
)
print("BTC 分析结果:", btc_analysis)
多因子特征工程实战
# features/multi_factor_builder.py
import pandas as pd
import numpy as np
from typing import Dict, List
import ccxt
from .sentiment import SentimentAnalyzer
class MultiFactorBuilder:
"""
多因子模型构建器
整合价格、链上、情绪等多维度数据
"""
def __init__(self, api_client=None):
self.exchange = ccxt.binance({'enableRateLimit': True})
self.sentiment_analyzer = SentimentAnalyzer(api_client)
def build_features(self, symbol: str, timeframe: str = '1h',
lookback_periods: int = 100) -> pd.DataFrame:
"""
构建完整的多因子特征矩阵
"""
# 1. 获取K线数据
ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, limit=lookback_periods)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
# 2. 技术因子
df = self._add_technical_factors(df)
# 3. 波动率因子
df = self._add_volatility_factors(df)
# 4. 资金流向因子
df = self._add_money_flow_factors(symbol, df)
# 5. 跨资产相关性因子
df = self._add_correlation_factors(symbol, df)
# 6. 情绪因子（调用 AI）
df = self._add_sentiment_factors(df)
return df.dropna()
def _add_technical_factors(self, df: pd.DataFrame) -> pd.DataFrame:
"""技术指标因子"""
# 移动平均线族
df['MA_5'] = df['close'].rolling(5).mean()
df['MA_20'] = df['close'].rolling(20).mean()
df['MA_60'] = df['close'].rolling(60).mean()
df['MA_ratio_5_20'] = df['MA_5'] / df['MA_20']
df['MA_ratio_20_60'] = df['MA_20'] / df['MA_60']
# 动量指标
df['RSI_14'] = self._calculate_rsi(df['close'], 14)
df['RSI_28'] = self._calculate_rsi(df['close'], 28)
df['MOM_12'] = df['close'].pct_change(12)
df['ROC_12'] = (df['close'] - df['close'].shift(12)) / df['close'].shift(12) * 100
# MACD
exp1 = df['close'].ewm(span=12, adjust=False).mean()
exp2 = df['close'].ewm(span=26, adjust=False).mean()
df['MACD'] = exp1 - exp2
df['MACD_signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
df['MACD_hist'] = df['MACD'] - df['MACD_signal']
# KD指标
low14 = df['low'].rolling(window=14).min()
high14 = df['high'].rolling(window=14).max()
df['RSV'] = (df['close'] - low14) / (high14 - low14) * 100
df['K'] = df['RSV'].ewm(com=2, adjust=False).mean()
df['D'] = df['K'].ewm(com=2, adjust=False).mean()
df['J'] = 3 * df['K'] - 2 * df['D']
# ADX 趋势强度
df['ADX'] = self._calculate_adx(df)
return df
def _add_volatility_factors(self, df: pd.DataFrame) -> pd.DataFrame:
"""波动率因子"""
# 历史波动率
df['HV_10'] = df['close'].pct_change().rolling(10).std() * np.sqrt(365)
df['HV_30'] = df['close'].pct_change().rolling(30).std() * np.sqrt(365)
# 布林带
df['BB_middle'] = df['close'].rolling(20).mean()
bb_std = df['close'].rolling(20).std()
df['BB_upper'] = df['BB_middle'] + 2 * bb_std
df['BB_lower'] = df['BB_middle'] - 2 * bb_std
df['BB_width'] = (df['BB_upper'] - df['BB_lower']) / df['BB_middle']
df['BB_position'] = (df['close'] - df['BB_lower']) / (df['BB_upper'] - df['BB_lower'])
# ATR
high_low = df['high'] - df['low']
high_close = np.abs(df['high'] - df['close'].shift())
low_close = np.abs(df['low'] - df['close'].shift())
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
df['ATR_14'] = tr.rolling(14).mean()
df['ATR_ratio'] = df['ATR_14'] / df['close']
return df
def _add_money_flow_factors(self, symbol: str, df: pd.DataFrame) -> pd.DataFrame:
"""资金流向因子"""
# 成交量变化率
df['volume_ma5'] = df['volume'].rolling(5).mean()
df['volume_ratio'] = df['volume'] / df['volume_ma5']
# OBV 能量潮
df['OBV'] = (np.sign(df['close'].diff()) * df['volume']).fillna(0).cumsum()
df['OBV_ma10'] = df['OBV'].rolling(10).mean()
df['OBV_slope'] = df['OBV'].pct_change()
# 价格与成交量背离
df['price_volume_corr'] = df['close'].rolling(20).corr(df['volume'])
return df
def _add_correlation_factors(self, symbol: str, df: pd.DataFrame) -> pd.DataFrame:
"""跨资产相关性因子"""
try:
# 获取主流币种数据计算相关性
base = symbol.split('/')[0]
corr_symbols = ['BTC/USDT' if base != 'BTC' else 'ETH/USDT']
for corr_sym in corr_symbols:
corr_ohlcv = self.exchange.fetch_ohlcv(corr_sym, '1h', limit=100)
corr_df = pd.DataFrame(corr_ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
corr_df['close'].index = df['timestamp'][:len(corr_df)]
df[f'corr_{corr_sym.replace("/", "_")}'] = df['close'].rolling(20).corr(
corr_df['close'].reindex(df.index, method='ffill')
)
except Exception:
df['corr_BTC_ETH'] = 0.5 # 默认值
return df
def _add_sentiment_factors(self, df: pd.DataFrame) -> pd.DataFrame:
"""情绪因子 - 通过 HolySheep AI 分析"""
# 使用最近N条新闻/推文更新情绪
if self.sentiment_analyzer:
sentiment = self.sentiment_analyzer.get_current_sentiment()
df['ai_sentiment'] = sentiment
df['sentiment_normalized'] = (sentiment + 100) / 200 # 归一化到 0-1
return df
def _calculate_rsi(self, prices: pd.Series, period: int) -> pd.Series:
"""计算 RSI"""
delta = prices.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
rs = gain / loss
return 100 - (100 / (1 + rs))
def _calculate_adx(self, df: pd.DataFrame, period: int = 14) -> pd.Series:
"""计算 ADX"""
high_low = df['high'] - df['low']
high_close = np.abs(df['high'] - df['close'].shift())
low_close = np.abs(df['low'] - df['close'].shift())
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
atr = tr.rolling(period).mean()
up_move = df['high'] - df['high'].shift()
down_move = df['low'].shift() - df['low']
plus_dm = np.where((up_move > down_move) & (up_move > 0), up_move, 0)
minus_dm = np.where((down_move > up_move) & (down_move > 0), down_move, 0)
plus_di = 100 * pd.Series(plus_dm).rolling(period).mean() / atr
minus_di = 100 * pd.Series(minus_dm).rolling(period).mean() / atr
dx = 100 * np.abs(plus_di - minus_di) / (plus_di + minus_di)
adx = dx.rolling(period).mean()
return adx
XGBoost 多因子模型训练与预测
# models/xgboost_quant_model.py
import pandas as pd
import numpy as np
from xgboost import XGBClassifier
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import classification_report, confusion_matrix
import pickle
from typing import Tuple, Dict
import os
class QuantXGBoostModel:
"""
XGBoost 多因子量化模型
目标：预测未来 N 个周期的涨跌方向
"""
def __init__(self, look_forward: int = 4, threshold: float = 0.002):
self.look_forward = look_forward # 预测未来 N 个周期
self.threshold = threshold # 涨跌阈值（过滤噪音）
self.model = None
self.feature_importance = None
self.scaler = None
def prepare_labels(self, df: pd.DataFrame) -> pd.Series:
"""
准备标签：未来收益 > threshold 为 1（多），< -threshold 为 0（空）
"""
future_return = df['close'].shift(-self.look_forward) / df['close'] - 1
labels = pd.Series(index=df.index, dtype=int)
labels[future_return > self.threshold] = 1 # 做多信号
labels[future_return < -self.threshold] = 0 # 做空信号
labels[abs(future_return) <= self.threshold] = 2 # 震荡/不操作
return labels
def train(self, df: pd.DataFrame, test_ratio: float = 0.2) -> Dict:
"""
训练 XGBoost 多因子模型
"""
# 分离特征和标签
labels = self.prepare_labels(df)
features = df.drop(columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
# 对齐索引
common_idx = features.index.intersection(labels.index)
X = features.loc[common_idx]
y = labels.loc[common_idx]
# 移除震荡标签，仅保留多空信号
mask = y != 2
X = X[mask]
y = y[mask]
# 时间序列分割（避免未来数据泄露）
split_idx = int(len(X) * (1 - test_ratio))
X_train, X_test = X[:split_idx], X[split_idx:]
y_train, y_test = y[:split_idx], y[split_idx:]
# 模型参数（根据我的实盘经验调优）
params = {
'n_estimators': 500,
'max_depth': 6,
'learning_rate': 0.03,
'subsample': 0.8,
'colsample_bytree': 0.8,
'min_child_weight': 5,
'gamma': 0.1,
'reg_alpha': 0.5,
'reg_lambda': 1.0,
'scale_pos_weight': len(y_train[y_train==0]) / len(y_train[y_train==1]),
'objective': 'binary:logistic',
'eval_metric': 'auc',
'use_label_encoder': False
}
self.model = XGBClassifier(**params)
self.model.fit(
X_train, y_train,
eval_set=[(X_test, y_test)],
verbose=50
)
# 保存特征重要性
self.feature_importance = pd.DataFrame({
'feature': X.columns,
'importance': self.model.feature_importances_
}).sort_values('importance', ascending=False)
# 评估
y_pred = self.model.predict(X_test)
report = classification_report(y_test, y_pred, output_dict=True)
return {
'classification_report': report,
'confusion_matrix': confusion_matrix(y_test, y_pred).tolist(),
'feature_importance': self.feature_importance.head(10).to_dict('records'),
'test_accuracy': report['accuracy']
}
def predict(self, df: pd.DataFrame) -> Tuple[np.ndarray, np.ndarray]:
"""
返回预测结果和概率
"""
features = df.drop(columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
features = features.dropna()
proba = self.model.predict_proba(features)
predictions = self.model.predict(features)
return predictions, proba
def get_trading_signal(self, df: pd.DataFrame,
confidence_threshold: float = 0.6) -> Dict:
"""
生成交易信号
结合 AI 分析进行二次确认
"""
predictions, proba = self.predict(df)
latest = df.iloc[-1]
signal = {
'timestamp': latest['timestamp'],
'close': latest['close'],
'direction': 'neutral',
'confidence': 0.0,
'position_size': 0.0,
'reason': ''
}
if len(proba) > 0:
last_proba = proba[-1]
pred = predictions[-1]
# 多头信号
if pred == 1 and last_proba[1] >= confidence_threshold:
signal['direction'] = 'long'
signal['confidence'] = float(last_proba[1])
signal['position_size'] = min(1.0, last_proba[1] ** 2) # 置信度平方调整仓位
signal['reason'] = f"模型置信度 {last_proba[1]:.2%}，XGBoost 多因子信号"
# 空头信号
elif pred == 0 and last_proba[0] >= confidence_threshold:
signal['direction'] = 'short'
signal['confidence'] = float(last_proba[0])
signal['position_size'] = min(1.0, last_proba[0] ** 2)
signal['reason'] = f"模型置信度 {last_proba[0]:.2%}，XGBoost 多因子信号"
return signal
def save(self, path: str = 'models/xgboost_quant.pkl'):
"""保存模型"""
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, 'wb') as f:
pickle.dump(self.model, f)
def load(self, path: str = 'models/xgboost_quant.pkl'):
"""加载模型"""
with open(path, 'rb') as f:
self.model = pickle.load(f)
完整回测系统
# backtest/backtest_engine.py
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple
from datetime import datetime
class BacktestEngine:
"""
向量化回测引擎
计算策略绩效指标
"""
def __init__(self, initial_capital: float = 100000,
commission_rate: float = 0.0004,
slippage: float = 0.0002):
self.initial_capital = initial_capital
self.commission_rate = commission_rate # 手续费（双向）
self.slippage = slippage # 滑点
def run(self, df: pd.DataFrame, signals: List[Dict]) -> Dict:
"""
执行回测
signals: [{"timestamp": ..., "direction": "long/short/neutral", "position_size": 0.0-1.0}]
"""
capital = self.initial_capital
position = 0 # 持仓数量
position_type = None # 'long' or 'short'
trades = []
equity_curve = [capital]
df_signals = pd.DataFrame(signals).set_index('timestamp')
for i, row in df.iterrows():
timestamp = row['timestamp']
close = row['close']
# 检查是否有信号
if timestamp in df_signals.index:
signal = df_signals.loc[timestamp]
# 平仓
if position != 0 and signal['direction'] == 'neutral':
pnl = self._close_position(position, position_type, close, capital)
capital += pnl
trades.append({
'entry_time': trades[-1]['entry_time'] if trades else timestamp,
'exit_time': timestamp,
'type': position_type,
'pnl': pnl,
'return': pnl / (position * trades[-1]['entry_price'] if trades else capital)
})
position = 0
position_type = None
# 开仓
elif position == 0 and signal['direction'] != 'neutral':
size = signal['position_size']
cost = capital * size
fees = cost * self.commission_rate * 2 # 进出双向
position = (cost - fees) / close
position_type = signal['direction']
trades.append({
'entry_time': timestamp,
'entry_price': close * (1 + self.slippage if signal['direction'] == 'long' else -self.slippage)
})
# 更新权益
if position != 0:
if position_type == 'long':
capital = position * close
else: # short
entry_price = trades[-1]['entry_price']
capital = self.initial_capital + (entry_price - close) * position
equity_curve.append(capital)
# 计算绩效指标
returns = pd.Series(equity_curve).pct_change().dropna()
metrics = self._calculate_metrics(returns, trades)
return {
'metrics': metrics,
'trades': trades,
'equity_curve': equity_curve,
'final_capital': capital
}
def _close_position(self, position: float, pos_type: str,
close_price: float, capital: float) -> float:
"""计算平仓盈亏"""
entry_price = close_price # 简化处理
if pos_type == 'long':
return position * close_price - capital
else:
return capital - position * close_price
def _calculate_metrics(self, returns: pd.Series, trades: List) -> Dict:
"""计算绩效指标"""
if len(returns) == 0:
return {}
total_return = (1 + returns).prod() - 1
annual_return = (1 + total_return) ** (252 * 24 / len(returns)) - 1
sharpe_ratio = returns.mean() / returns.std() * np.sqrt(252 * 24)
# 最大回撤
cumulative = (1 + returns).cumprod()
running_max = cumulative.expanding().max()
drawdown = (cumulative - running_max) / running_max
max_drawdown = drawdown.min()
# 胜率
winning_trades = [t['pnl'] for t in trades if t.get('pnl', 0) > 0]
win_rate = len(winning_trades) / len(trades) if trades else 0
return {
'total_return': f"{total_return:.2%}",
'annual_return': f"{annual_return:.2%}",
'sharpe_ratio': f"{sharpe_ratio:.2f}",
'max_drawdown': f"{max_drawdown:.2%}",
'win_rate': f"{win_rate:.2%}",
'total_trades': len(trades),
'profit_factor': abs(sum([t['pnl'] for t in trades if t.get('pnl', 0) > 0]) /
sum([t['pnl'] for t in trades if t.get('pnl', 0) < 0])) if trades else 0
}
主程序入口
# main.py
import os
from dotenv import load_dotenv
from api.holysheep_client import HolySheepAIClient
from features.multi_factor_builder import MultiFactorBuilder
from models.xgboost_quant_model import QuantXGBoostModel
from backtest.backtest_engine import BacktestEngine
load_dotenv()
def main():
# 初始化 HolySheep API 客户端
client = HolySheepAIClient(api_key=os.getenv("HOLYSHEEP_API_KEY"))
# 初始化模块
factor_builder = MultiFactorBuilder(api_client=client)
model = QuantXGBoostModel(look_forward=4, threshold=0.003)
print("=" * 60)
print("加密货币量化策略：机器学习多因子模型")
print("=" * 60)
# 1. 构建特征
print("\n[1/4] 构建多因子特征矩阵...")
df = factor_builder.build_features('BTC/USDT', timeframe='1h', lookback_periods=500)
print(f"特征数量: {len(df.columns)}，数据长度: {len(df)}")
# 2. 训练模型
print("\n[2/4] 训练 XGBoost 多因子模型...")
train_results = model.train(df, test_ratio=0.2)
print(f"测试集准确率: {train_results['test_accuracy']:.2%}")
print("\nTop 10 重要因子:")
for i, feat in enumerate(train_results['feature_importance'][:10], 1):
print(f" {i}. {feat['feature']}: {feat['importance']:.4f}")
# 3. 生成交易信号
print("\n[3/4] 生成交易信号...")
signal = model.get_trading_signal(df, confidence_threshold=0.65)
print(f"方向: {signal['direction']}")
print(f"置信度: {signal['confidence']:.2%}")
print(f"建议仓位: {signal['position_size']:.0%}")
print(f"理由: {signal['reason']}")
# 4. 使用 AI 进行二次确认
print("\n[4/4] AI 信号二次确认...")
try:
indicators = {
'RSI_14': float(df['RSI_14'].iloc[-1]),
'MACD': float(df['MACD'].iloc[-1]),
'ADX': float(df['ADX'].iloc[-1]) if 'ADX' in df.columns else 50,
'BB_position': float(df['BB_position'].iloc[-1]) if 'BB_position' in df.columns else 0.5
}
ai_analysis = client.generate_market_analysis(
symbol='BTC/USDT