比特币自动交易系统开发指南

1。开发环境的组成

# Install required libraries
pip install ccxt pandas numpy talib TA-Lib python-dotenv schedule websockets flask

• 主要工具：
  • CCXT：100+交换集成API
  • 熊猫：数据分析
  • TA-LIB：技术指标的计算
  • WebSocket：接收真实 - 时间数据
  • 烧瓶：仪表板结构

2。选择Exchange并设置API

# Save API key in .env file
API_KEY = "your_api_key" # API key issued by exchange
API_SECRET = "your_api_secret" # API secret issued by exchange

• 推荐的交换：
  • Binance：高流动性，REST/WEBSOCKENT支持
  • Bybit：礼品交易的专业化
  • 乐观：国内子公司稳定

3。图表数据的收集

import ccxt

# Initialize Binance API
binance = ccxt.binance({
    'apiKey': API_KEY,  # Set API key
    'secret': API_SECRET,  # Set API secret
    'enableRateLimit': True  # Enable API request limit
})

# Receive real-time OHLCV data (Websocket)
async def fetch_btc_data():
    async with websockets.connect('wss://fstream.binance.com/ws/btcusdt@kline_1m') as ws:
        while True:
            data = await ws.recv()  # Receive real-time data
            print(json.loads(data))  # Print data

4。交易策略制定

4-1。 实施技术指标

import talib

# Function to calculate technical indicators
def calculate_indicators(df):
    # Calculate 20-day Simple Moving Average (SMA)
    df['MA20'] = talib.SMA(df['close'], timeperiod=20)
    # Calculate 14-day Relative Strength Index (RSI)
    df['RSI'] = talib.RSI(df['close'], timeperiod=14)
    # Calculate MACD (Moving Average Convergence Divergence)
    df['MACD'], _, _ = talib.MACD(df['close'])
    return df

4-2。 销售信号创建逻辑

# Function to generate trading signals
def generate_signal(df):
    latest = df.iloc[-1]  # Get the latest data
    
    # Dual SMA Strategy
    if latest['MA20'] > latest['MA50'] and df['MA20'].iloc[-2] <= df['MA50'].iloc[-2]:
        return 'BUY'  # Buy signal
    elif latest['MA20'] < latest['MA50'] and df['MA20'].iloc[-2] >= df['MA50'].iloc[-2]:
        return 'SELL'  # Sell signal
    else:
        return 'HOLD'  # Hold signal

5。风险管理系统

# Function to calculate position size
def calculate_position_size(balance, risk_per_trade=0.02):
    # Calculate position size based on total capital and risk percentage per trade
    return balance * risk_per_trade

# Function to set stop loss
def set_stop_loss(entry_price, atr, multiplier=1.5):
    # Calculate stop loss based on ATR (Average True Range)
    return entry_price - (atr * multiplier)

6。订单执行模块

# Function to execute an order
def execute_order(side, amount, symbol='BTC/USDT'):
    try:
        if side == 'BUY':
            # Execute a market buy order
            order = binance.create_market_buy_order(symbol, amount)
        elif side == 'SELL':
            # Execute a market sell order
            order = binance.create_market_sell_order(symbol, amount)
        print(f"Order Executed: {order}")  # Print order execution log
        return order
    except Exception as e:
        print(f"Order Failed: {e}")  # Print error message if the order fails
        return None

7。回测系统

# Backtesting function
def backtest_strategy(df, initial_balance=10000):
    balance = initial_balance  # Set initial capital
    position = 0  # Initialize position
    
    for i in range(1, len(df)):
        signal = df['signal'].iloc[i]  # Get trading signal
        price = df['close'].iloc[i]  # Get closing price
        
        if signal == 'BUY' and position == 0:
            # Set position when a buy signal is received
            position = balance / price
            balance = 0
        elif signal == 'SELL' and position > 0:
            # Liquidate position when a sell signal is received
            balance = position * price
            position = 0
            
    return balance  # Return final balance

8。监视和记录系统

import logging

# Logging settings
logging.basicConfig(filename='trading.log', level=logging.INFO)

# Transaction logging function
def log_transaction(order):
    logging.info(f"""
    [Transaction Details]
    Time: {datetime.now()}
    Type: {order['side']}
    Amount: {order['amount']}
    Price: {order['price']}
    Status: {order['status']}
    """)

9。主要功能

# Main execution loop
def main():
    while True:
        try:
            df = fetch_realtime_data()  # Fetch real-time data
            df = calculate_indicators(df)  # Calculate technical indicators
            signal = generate_signal(df)  # Generate trading signal
            
            if signal != 'HOLD':
                # Execute an order if there is a trading signal
                amount = calculate_position_size(get_balance())
                execute_order(signal, amount)
                
            time.sleep(60)  # Run at 1-minute intervals
        except KeyboardInterrupt:
            break  # Exit the loop when interrupted by the user

10。安全增强措施

10-1。 API密钥管理

• 切勿提交版本管理系统
• 使用AWS Secrets Manager或Hashicorp Vault。

10-2

binance = ccxt.binance({
    'options': {'adjustForTimeDifference': True},  # Time difference correction
    'proxies': {'https': 'http://10.10.1.10:3128'}  # Proxy settings
})

11。分销示例

# System service registration (Linux)
[Unit]
Description=Crypto Trading Bot
After=network.target

[Service]
ExecStart=/usr/bin/python3 /path/to/bot.py
Restart=always

[Install]
WantedBy=multi-user.target

📌核心预防措施

• 限制不到初始资本的1％的风险
• 必须观察到Exchange API率限制
• 准备周末/假日市场波动
• 定期重新评估战略绩效
• 在实际资金之前，请确保在虚拟环境中测试超过2周
• 可以仅执行要发行和应用的实际API地址