加密货币量化交易代码

import ccxt.async_support as ccxt
import pandas as pd
import asyncio
import os
import time
from datetime import datetime
from dotenv import load_dotenv

# 加载配置
load_dotenv()

# ================= 配置区域 (在此修改参数) =================
SYMBOL = ‘BTC/USDC:USDC‘   # 交易对
TIMEFRAME = ‘1m‘   # K 线周期
LEVERAGE = 50   # 杠杆倍数
POSITION_SIZE_PCT = 0.20   # 每次开仓占用余额比例 (20%)
ORDER_TIMEOUT_SEC = 10   # 挂单超时时间 (秒)

# 【新增】止盈止损设置 (单位：百分比)
STOP_LOSS_PCT = 2.0   # 亏损 2% 自动止损
TAKE_PROFIT_PCT = 5.0   # 盈利 5% 自动止盈

# 策略指标参数
RSI_PERIOD = 14
BB_PERIOD = 20
BB_STD = 2.0
EMA_FAST = 12
EMA_SLOW = 26
# =========================================================

class TradingBot:
   def __init__(self):
   self.api_key = os.getenv(‘BINANCE_API_KEY‘)
   self.secret = os.getenv(‘BINANCE_SECRET_KEY‘)
   is_testnet = os.getenv(‘IS_TESTNET‘, ‘false‘).lower() == ‘true‘
  
   if not self.api_key or not self.secret:
   print("❌ 错误：未在 .env 文件中找到 API 密钥！")
   exit(1)

   self.exchange = ccxt.binanceusdm({
   ‘apiKey‘: self.api_key,
   ‘secret‘: self.secret,
   ‘enableRateLimit‘: True,
   ‘options‘: {‘defaultType‘: ‘future‘}
   })
  
   if is_testnet:
   self.exchange.set_sandbox_mode(True)
   print("⚠️ 当前运行在【测试网】模式")
   else:
   print("⚠️ 警告：当前运行在【实盘】模式！请确保资金安全！")

   async def init_market(self):
   try:
   await self.exchange.load_markets()
   await self.exchange.set_leverage(LEVERAGE, SYMBOL)
   print(f"✅ {SYMBOL} 杠杆已设置为 {LEVERAGE}x")
  
   balance = await self.exchange.fetch_balance()
   usdc_balance = balance[‘total‘].get(‘USDC‘, 0)
   print(f" 当前 USDC 余额：{usdc_balance}")
   return usdc_balance
   except Exception as e:
   print(f"❌ 初始化失败：{e}")
   return None

   async def fetch_data(self):
   try:
   ohlcv = await self.exchange.fetch_ohlcv(SYMBOL, TIMEFRAME, limit=100)
   df = pd.DataFrame(ohlcv, columns=[‘time‘, ‘open‘, ‘high‘, ‘low‘, ‘close‘, ‘volume‘])
  
   # --- 手动计算指标 ---
   # RSI
   delta = df[‘close‘].diff()
   gain = (delta.where(delta > 0, 0)).rolling(window=RSI_PERIOD).mean()
   loss = (-delta.where(delta ema_s:
   return ‘LONG‘
  
   # 做空信号：RSI 超买 + 触及上轨 + 快线在慢线下方
   if rsi > 65 and close >= bbu * 0.999 and ema_f ORDER_TIMEOUT_SEC:
   print(f"⏰ 超时 ({ORDER_TIMEOUT_SEC}s)，取消订单...")
   try:
   await self.exchange.cancel_order(order_id, SYMBOL)
   print("❌ 订单已取消")
   return False
   except:
   return False
  
   status_order = await self.exchange.fetch_order(order_id, SYMBOL)
   if status_order[‘status‘] == ‘closed‘:
   print(f"✅ 订单已完全成交！成交价：{status_order[‘average‘]}")
   return True
   elif status_order[‘status‘] in [‘canceled‘, ‘rejected‘]:
   return False
  
   await asyncio.sleep(1)
  
   except Exception as e:
   print(f"❌ 下单出错：{e}")
   return False

   async def monitor_position(self, entry_price, amount, position_side):
   """监控持仓，实现止盈止损"""
   print(f"️ 启动监控：入场价 {entry_price} | 止损 {-STOP_LOSS_PCT}% | 止盈 +{TAKE_PROFIT_PCT}%")
  
   close_side = ‘sell‘ if position_side == ‘LONG‘ else ‘buy‘
  
   while True:
   try:
   ticker = await self.exchange.fetch_ticker(SYMBOL)
   current_price = ticker[‘last‘]
  
   # 计算盈亏百分比
   if position_side == ‘LONG‘:
   pnl_pct = (current_price - entry_price) / entry_price * 100
   else:
   pnl_pct = (entry_price - current_price) / entry_price * 100
  
   # 每 5 秒打印一次状态
   if int(time.time()) % 5 == 0:
   print(f" 监控中：现价 {current_price:.2f} | 盈亏 {pnl_pct:.2f}%")

   # 触发止损
   if pnl_pct = TAKE_PROFIT_PCT:
   print(f" 触发止盈！盈利 {pnl_pct:.2f}%，市价平仓...")
   await self.exchange.create_market_order(SYMBOL, close_side, amount)
   print("✅ 止盈平仓完成！")
   return True # 退出监控
  
   await asyncio.sleep(2) # 每 2 秒检查一次
  
   except Exception as e:
   print(f"⚠️ 监控出错：{e}")
   await asyncio.sleep(5)

   async def run_loop(self):
   await self.init_market()
  
   while True:
   try:
   print(f"\n⏰ [{datetime.now().strftime(‘%H:%M:%S‘)}] 扫描市场...")
  
   data = await self.fetch_data()
   if data is None:
   await asyncio.sleep(5)
   continue
  
   signal = self.generate_signal(data)
  
   if signal:
   print(f" 发现信号：{signal}")
  
   balance_info = await self.exchange.fetch_balance()
   free_usdc = balance_info[‘free‘].get(‘USDC‘, 0)
  
   if free_usdc < 10:
   print(" 余额不足，跳过")
   await asyncio.sleep(60)
   continue

   # 计算仓位
   margin_amount = free_usdc * POSITION_SIZE_PCT
   target_notional = margin_amount * LEVERAGE
   amount = target_notional / data[‘close‘]
   amount = float(f"{amount:.4f}") # 保留 4 位小数
  
   ticker = await self.exchange.fetch_ticker(SYMBOL)
   if signal == ‘LONG‘:
   price = ticker[‘bid‘] * 1.0002
   side = ‘buy‘
   pos_side = ‘LONG‘
   else:
   price = ticker[‘ask‘] * 0.9998
   side = ‘sell‘
   pos_side = ‘ORT‘
  
   success = await self.place_order_with_timeout(side, price, amount)
  
   if success:
   # === 关键：进入止盈止损监控 ===
   await self.monitor_position(price, amount, pos_side)
  
   print("✨ 本轮交易结束，冷却 60 秒...")
   await asyncio.sleep(60)
   else:
   print("⚠️ 挂单未成交，继续监测...")
  
   else:
   print(" 无明确信号，等待下一根 K 线...")
  
   await asyncio.sleep(55)

   except Exception as e:
   print(f" 主循环严重错误：{e}")
   await asyncio.sleep(10)

   async def close(self):
   await self.exchange.close()

if __name__ == ‘__main__‘:
   bot = TradingBot()
   try:
   asyncio.run(bot.run_loop())
   except KeyboardInterrupt:
   print("\n 用户手动中断")
   finally:
   asyncio.run(bot.close())