pip install ccxt pandas numpy TA-Lib python-dotenv
import ccxt
import pandas as pd
binance = ccxt.binance({
'apiKey': 'YOUR_API_KEY',
'secret': 'YOUR_SECRET',
'enableRateLimit': True
})
<br/>
# 查询 1 小时柱状图数据
ohlcv = binance.fetch_ohlcv('BTC/USDT', '1h')
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
import talib
<br/>
# RSI (14 周期)
df['RSI'] = talib.RSI(df['close'], timeperiod=14)
<br/>
# MACD (12,26,9)
df['MACD'], df['MACD_signal'], _ = talib.MACD(df['close'],
fastperiod=12,
slowperiod=26,
signalperiod=9)
<br/>
# 50 日 & 200 日移动平均线
df['MA50'] = talib.SMA(df['close'], timeperiod=50)
df['MA200'] = talib.SMA(df['close'], timeperiod=200)
<br/>
# 布林带 (20 日)
df['upper_band'], df['middle_band'], df['lower_band'] = talib.BBANDS(df['close'],
timeperiod=20)
def generate_signal(row):
# 上行动量条件
bull_condition = (
(row['RSI'] > 50) &
(row['MACD'] > row['MACD_signal']) &
(row['close'] > row['MA200']) &
(row['close'] > row['upper_band'])
# 下行动量条件
bear_condition = (
(row['RSI'] < 45) &
(row['MACD'] < row['MACD_signal']) &
(row['close'] < row['MA200']) &
(row['close'] < row['lower_band'])
if bull_condition:
return 'BUY'
elif bear_condition:
return 'SELL'
else:
return 'HOLD'
df['signal'] = df.apply(generate_signal, axis=1)
def calculate_position_size(balance, risk_percent=2, stop_loss_pct=5):
max_risk = balance * (risk_percent/100)
return max_risk / (stop_loss_pct/100)
def dynamic_stoploss(current_price, atr, multiplier=2):
return current_price - (atr * multiplier)
def update_trailing_stop(high_price, trail_percent=3):
return high_price * (1 - (trail_percent/100))
initial_balance = 10000 # USDT
position = 0
balance = initial_balance
trailing_stop = None
for idx, row in df.iterrows():
current_price = row['close']
if row['signal'] == 'BUY' and position == 0:
position = balance / current_price
balance = 0
trailing_stop = current_price * 0.97 # 3% 初始止损
elif row['signal'] == 'SELL' or current_price < trailing_stop:
if position > 0:
balance = position * current_price
position = 0
trailing_stop = None
elif position > 0:
# 追踪止损更新
trailing_stop = max(trailing_stop, current_price * 0.97)
final_balance = balance + (position * df.iloc[-1]['close'])
print(f"最终回报率: {((final_balance/initial_balance)-1)*100:.2f}%")
❗ 禁止过度拟合: 需要使用 3 年以上的数据进行验证
❗ 确认流动性: 避免交易量低于 100 BTC 的区间
❗ 限制自动化: 严格遵守 API 调用次数(通常为每分钟 120 次)
实际应用前务必进行 6 个月以上的模拟投资
0