写了个交易机器人：双均线策略的实盘记录

import ccxt
import pandas as pd
import time
from datetime import datetime

exchange = ccxt.okx({
    'apiKey': 'YOUR_API_KEY',
    'secret': 'YOUR_SECRET_KEY',
    'password': 'YOUR_API_PASSPHRASE',
    'enableRateLimit': True,
})

symbol = 'BTC-USDT-SWAP'
timeframe = '1m'
limit = 50
position_size = 50  # USDT
leverage = 3

def get_ma_crossover():
    """获取最新两根K线，计算MA5和MA20，判断是否金叉/死叉"""
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
    df['ma5'] = df['close'].rolling(window=5).mean()
    df['ma20'] = df['close'].rolling(window=20).mean()
    
    # 获取倒数第二根（上一根已收盘的K线）和倒数第三根的金叉状态
    prev_ma5 = df['ma5'].iloc[-3]
    prev_ma20 = df['ma20'].iloc[-3]
    curr_ma5 = df['ma5'].iloc[-2]
    curr_ma20 = df['ma20'].iloc[-2]
    
    # 金叉：之前MA5 <= MA20，现在MA5 > MA20
    golden_cross = prev_ma5 <= prev_ma20 and curr_ma5 > curr_ma20
    # 死叉：之前MA5 >= MA20，现在MA5 < MA20
    death_cross = prev_ma5 >= prev_ma20 and curr_ma5 < curr_ma20
    
    return golden_cross, death_cross

def get_position():
    """查询当前持仓"""
    positions = exchange.fetch_positions([symbol])
    for pos in positions:
        if pos['symbol'] == symbol and float(pos['contracts']) > 0:
            return pos
    return None

def place_order(side, amount):
    """下单，带止盈止损"""
    # 先设置杠杆
    exchange.set_leverage(leverage, symbol)
    
    # 开仓
    order = exchange.create_market_order(
        symbol, side, amount, 
        params={'leverage': leverage}
    )
    
    # 获取开仓均价，设置止盈止损
    # 这里需要根据实际成交价来设置
    # 实际项目中会从 order 或后续查询中获取 fill price
    print(f"{datetime.now()} | {'多头' if side == 'buy' else '空头'}开仓 | 数量: {amount}")
    return order

def main_loop():
    """主循环"""
    print(f"策略启动: {symbol} | 时间周期: {timeframe} | 杠杆: {leverage}x")
    
    while True:
        try:
            golden, death = get_ma_crossover()
            position = get_position()
            
            if golden and not position:
                # 金叉且无持仓 → 开多
                place_order('buy', position_size)
            elif death and position:
                # 死叉且有持仓 → 平多（可改为开空，根据策略偏好）
                exchange.create_market_order(symbol, 'sell', position['contracts'])
                print(f"{datetime.now()} | 平仓")
            
            time.sleep(10)  # 每10秒检查一次
        except Exception as e:
            print(f"错误: {e}")
            time.sleep(30)

if __name__ == '__main__':
    main_loop()

# 资金管理模块示例
def calculate_position_size(balance, risk_per_trade=0.02, stop_loss_pct=0.0015):
    """
    根据账户余额、单笔风险和止损比例计算仓位
    balance: 账户余额 (USDT)
    risk_per_trade: 单笔最大亏损比例 (2%)
    stop_loss_pct: 止损比例 (0.15%)
    """
    max_loss = balance * risk_per_trade
    position = max_loss / stop_loss_pct
    return min(position, balance)  # 不超过总余额

[Unit]
Description=okx-bot
After=network.target

[Service]
ExecStart=/usr/bin/python3 /opt/okx-bot/bot.py
Restart=always
RestartSec=10
User=root
WorkingDirectory=/opt/okx-bot
StandardOutput=append:/var/log/okx-bot.log
StandardError=append:/var/log/okx-bot.log

[Install]
WantedBy=multi-user.target

sudo cp okx.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable okx-bot
sudo systemctl start okx-bot
# 查看状态
sudo systemctl status okx-bot
# 查看日志
journalctl -u okx-bot -f

# config.yaml
exchange:
  name: okx
  api_key: "YOUR_API_KEY"
  api_secret: "YOUR_SECRET"
  password: "YOUR_PASSPHRASE"
  sandbox: false  # 先开沙箱测试

strategy:
  symbol: "BTC-USDT-SWAP"
  timeframe: "1m"
  ma_fast: 5
  ma_slow: 20
  leverage: 3
  position_size: 50  # USDT
  take_profit_pct: 0.3
  stop_loss_pct: 0.15
  check_interval: 10  # 检查间隔（秒）

telegram:
  bot_token: "YOUR_BOT_TOKEN"
  chat_id: "YOUR_CHAT_ID"

exchange = ccxt.okx({
    'apiKey': 'SANDBOX_API_KEY',
    'secret': 'SANDBOX_SECRET',
    'password': 'SANDBOX_PASSPHRASE',
    'enableRateLimit': True,
    'urls': {
        'api': {
            'public': 'https://www.okx.com/api/v5/public',
            'private': 'https://www.okx.com/api/v5/private'
        }
    }
})

import requests

class TelegramNotifier:
    def __init__(self, bot_token, chat_id):
        self.base_url = f"https://api.telegram.org/bot{bot_token}"
        self.chat_id = chat_id
    
    def send_message(self, text):
        url = f"{self.base_url}/sendMessage"
        data = {
            'chat_id': self.chat_id,
            'text': text,
            'parse_mode': 'HTML'
        }
        try:
            requests.post(url, data=data, timeout=10)
        except Exception as e:
            print(f"Telegram 通知失败: {e}")

notifier = TelegramNotifier(config['telegram']['bot_token'], config['telegram']['chat_id'])

# 使用时
notifier.send_message(
    f"🤖 <b>开多</b>\n"
    f"价格: {current_price}\n"
    f"数量: {position_size} USDT\n"
    f"杠杆: {leverage}x\n"
    f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
)

def track_slippage(expected_price, filled_price, side):
    slippage = abs(filled_price - expected_price) / expected_price * 100
    # 记录到 CSV，定期分析
    with open('slippage_log.csv', 'a') as f:
        f.write(f"{datetime.now()},{side},{expected_price},{filled_price},{slippage:.4f}%\n")

def check_latency():
    """检查 API 延迟，如果太高就告警"""
    start = time.time()
    exchange.fetch_ticker(symbol)
    latency = (time.time() - start) * 1000  # 毫秒
    if latency > 1000:
        notifier.send_message(f"⚠️ 延迟告警: {latency:.0f}ms")
    return latency

def should_skip_trade(df, volatility_threshold=0.005):
    """如果最近 N 根 K 线的波动太小，跳过交易避免震荡磨损"""
    recent_high = df['high'].iloc[-10:].max()
    recent_low = df['low'].iloc[-10:].min()
    volatility = (recent_high - recent_low) / recent_low
    if volatility < volatility_threshold:
        return True  # 波动太小，跳过
    return False