【python数字货币量化_21精华帖_策略第9篇】永续合约、交割合约client_oid版本代码迭代1.0

显示全部楼层 · 2023-8-2 13:46:57

这次迭代主要基于一个问题考虑，就是reblance的问题，短线在开单长期磨损的情况下，可能会磨损掉长线策略资金。这次将短线策略集中于永续合约，长线策略集中于交割合约。代码已提供，还有很多可精简的地方。另外该策略未使用并行模式，个人觉得，多开几个策略和并行应该是差不多的效果。
永续合约版本
"""
# 程序主要思路如下：
通过while语句，每根K线不断的循环。
每次循环中需要做的操作步骤
1. 更新账户信息
2. 获取实时数据
3. 根据最新数据计算买卖信号
4. 根据目前仓位、买卖信息，结束本次循环，或者进行交易
5. 交易
"""
from urllib.request import urlopen
import ccxt
import time
import pandas as pd
from datetime import datetime, timedelta
from time import sleep
import json
import warnings
import requests
import random
import hashlib
import hmac
import base64
from urllib import parse
import numpy as np
import Signal
warnings.filterwarnings("ignore")
pd.set_option('expand_frame_repr', False)  # 当列太多时不换行
pd.set_option("display.max_columns",100)

coin="BTC-USDT"
strategy = "vix"

config =pd.read_csv('swap_usdt.csv',index_col="coin")
exchange = ccxt.okex3()
exchange.load_markets()
exchange.apiKey = config["apiKey"][coin]
exchange.secret = config["secret"][coin]
exchange.password=config["password"][coin]
symbol =coin + '-SWAP'
time_interval = config["time_interval"][coin]
trade_coin = symbol.split("-")[0]
base_coin = symbol.split("-")[1]
secret = config["robot_secret"][coin]
robot_id = config["robot_id"][coin]

def send_dingding_msg(content, secret=secret, robot_id=robot_id):
timestamp = int(round(time.time() * 1000))
secret_enc = bytes(secret.encode('utf-8'))
string_to_sign = '{}\n{}'.format(timestamp, secret)
string_to_sign_enc = bytes(string_to_sign.encode('utf-8'))
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = parse.quote_plus(base64.b64encode(hmac_code))
query_string = 'access_token=' + str(robot_id) + "&timestamp=" + str(timestamp) + "&sign=" + str(sign)
try:
      msg = {
         "msgtype": "text",
         "text": {"content": content + "\n" + datetime.now().strftime("%Y-%m-%d %H:%M:%S")}}
      url = 'https://oapi.dingtalk.com/robot/send'
      url = url + '?' + query_string
      body = json.dumps(msg)
      # 设置重连次数
      requests.adapters.DEFAULT_RETRIES = 15
      # 设置连接活跃状态为False
      s = requests.session()
      s.keep_alive = False
      #  进行请求
      requests.post(url, data=body, headers={"Content-Type": "application/json ;charset=utf-8"}, timeout=5)
      print('成功发送钉钉')
except Exception as e:
      print('钉钉发送失败,尝试重复发送', e)
      '''请求超时'''
      for i in range(1, 5):
         print('请求超时，第%s次重复请求' % i)
         response = requests.post(url, data=body, headers={"Content-Type": "application/json ;charset=utf-8"},
                                 timeout=5)
         requests.adapters.DEFAULT_RETRIES = 5
         if response.status_code == 200:
            print('成功发送钉钉')
            return response

def place_order(exchange, symbol,client_oid,order_type, buy_or_sell, price, match_price, size):
"""
下单
:param lever_rate:
:param exchange:
:param symbol:
:param order_type:
:param buy_or_sell: 1:开多 2:开空 3:平多 4:平空
:param price:
:param amount: 这里的amount是合约张数
:return:
"""
if order_type == '1':
      match_price = '1'
elif order_type == '0':
      match_price = '0'
if buy_or_sell == '开多':
      type = '1'
elif buy_or_sell == '开空':
      type = '2'
elif buy_or_sell == '平多':
      type = '3'
elif buy_or_sell == '平空':
      type = '4'
# 下单
for i in range(20):
      try:
         order_info = exchange.swap_post_order(params={'client_oid': client_oid,
                                                         'type': type,
                                                         'instrument_id': symbol,
                                                         'order_type': 0,
                                                         'price': price,
                                                         'size': size,
                                                         'match_price': match_price,
                                                         'leverage': 3
                                                      } )
         print('okex 合约交易下单成功：', symbol,  buy_or_sell, price, size)
         print('下单信息：', order_info, '\n')
         return order_info

      except Exception as e:
         if "error_code" in str(e) and "35014" in str(e):
            _df = exchange.swap_get_instruments_instrument_id_price_limit(params={'instrument_id': symbol})
            limit_high = float(_df["highest"])
            limit_low = float(_df["lowest"])
            print("限价数据：", limit_high, limit_low, "下单价格：", price)
            if buy_or_sell in ["开多", "平空"]:
                  price = limit_high * 0.995
            else:
                  price = limit_low * 1.005
            print("修改后的下单价格：", price)
            time.sleep(0.5)
         if "error_code" in str(e) and "35002" in str(e):
            time.sleep(15)
         if "error_code" in str(e) and "35004" in str(e):
            time.sleep(15)
         print(e)
raise ValueError('okex 合约交易下单报错次数过多，程序终止')
print('下单报错次数过多，程序终止')
send_dingding_msg('下单报错次数过多，程序终止')

def fetch_account_info(exchange,symbol,client_oid):
"""
获取账户信息
:param exchange: 交易所
:return:
:param account_info: dict形式，包含账户和仓位信息
"""
# 创建用于存放账户信息的变量
account_info = {'账户余额': 0, '多头仓位数量': 0, '空头仓位数量': 0, '仓位成本': 0}
while True:
   # 获取账户的资产信息
      try:
         data = exchange.swap_get_instrument_id_accounts(params={'instrument_id':symbol})  # 从交易所获取账户balance信息
      except Exception as e:
         send_dingding_msg('获取账户信息失败')
         print(e)
         sleep(0.2)
         continue
      # 将数据转化为df格式
      account_info['账户余额'] = float(data['info']["equity"])

      # 获取账户的持仓信息
      try:
         client_oid_info = exchange.swap_get_orders_instrument_id_client_oid(
            params={'instrument_id': symbol, "client_oid": client_oid})
      except Exception as e:
         try:
            if "error_code" in str(e) and "35029" in str(e):
                  account_info['空头仓位数量'] = 0
                  account_info['多头仓位数量'] = 0
                  account_info['仓位成本'] = 0
                  break
            time.sleep(0.05)
         except:
            time.sleep(0.05)
         continue

      if client_oid_info:  # 当持仓信息信息不为空时
         if client_oid_info['state'] == "2":
            if client_oid_info['type'] == "1":  # 当持仓信息信息不为空时
                  account_info['多头仓位数量'] = float(client_oid_info['filled_qty'])
                  account_info['仓位成本'] = float(client_oid_info['price_avg'])
            elif client_oid_info['type'] == "2":  # 当持仓信息信息不为空时
                  account_info['空头仓位数量'] = float(client_oid_info['filled_qty'])
                  account_info['仓位成本'] = float(client_oid_info['price_avg'])
            elif client_oid_info['type'] == "3" or "4":  # 当持仓信息信息不为空时
                  account_info['空头仓位数量'] = 0
                  account_info['多头仓位数量'] = 0
                  account_info['仓位成本'] = 0
      break
return account_info

def get_history_candle_data(exchange, symbol, time_interval, max_len, max_try_amount=20):
"""
获取某个币种在okex交易所所有能获取的历史数据，目前v3接口最多获取1440根
:param exchange:
:param symbol:
:param time_interval_str:
:param now_milliseconds:
:param max_len:
:param max_try_amount:
:return:
"""
now_milliseconds = int(time.time() * 1e3)  # 获取统一的数据截止时间
time_interval_int = int(time_interval[:-1])
time_segment = time_interval_int * 60 * 1000
end = now_milliseconds - time_segment
since = end - max_len * time_segment
all_kline_data = []
while end - since >= time_segment:
      kline_data = []
      for i in range(max_try_amount):
         try:
            kline_data = exchange.fetch_ohlcv(symbol=symbol, since=since, timeframe=time_interval)
            break
         except Exception as e:
            print(e)
            try:
                  e_str = str(e)
                  if '35004' in e_str:
                     time.sleep(30)
                  if '30014' in e_str:
                     time.sleep(10)
            except:
                  time.sleep(10)

            if i == (max_try_amount - 1):
                  print('获取历史数据时，fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')
                  send_dingding_msg('获取历史数据时，fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')
                  raise ValueError('获取历史数据时，fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')
      if kline_data:
         # 更新获取时间
         since = kline_data[-1][0]
         all_kline_data += kline_data
      else:
         break

# 对数据进行整理
df = pd.DataFrame(all_kline_data, dtype=float)
df.rename(columns={0: 'MTS', 1: 'open', 2: 'high', 3: 'low', 4: 'close', 5: 'volume'}, inplace=True)
df['candle_begin_time'] = pd.to_datetime(df['MTS'], unit='ms')
df['candle_begin_time_GMT8'] = df['candle_begin_time'] + timedelta(hours=8)
df = df[['candle_begin_time_GMT8', 'open', 'high', 'low', 'close', 'volume']]

# 删除重复的数据
df.drop_duplicates(subset=['candle_begin_time_GMT8'], keep='last', inplace=True)
df.reset_index(drop=True, inplace=True)
df = df[:-1]

return df

def ccxt_fetch_candle_data(exchange, symbol, time_interval, limit, max_try_amount=5):
"""
:param exchange:
:param symbol:
:param time_interval:
:param limit:
:param max_try_amount:
:return:
获取K线数据
"""
for _ in range(max_try_amount):
      try:
         # 获取数据
         data = exchange.fetch_ohlcv(symbol=symbol, timeframe=time_interval, limit=limit)
         # 整理数据
         df = pd.DataFrame(data, dtype=float)
         df.rename(columns={0: 'MTS', 1: 'open', 2: 'high',
                           3: 'low', 4: 'close', 5: 'volume'}, inplace=True)
         df['candle_begin_time'] = pd.to_datetime(df['MTS'], unit='ms')
         df['candle_begin_time_GMT8'] = df['candle_begin_time'] + timedelta(hours=8)
         df = df[['candle_begin_time_GMT8', 'open', 'high', 'low', 'close', 'volume']]
         return df
      except Exception as e:
         print(e)
         print('获取fetch_ohlcv合约K线数据失败，稍后重试')
         time.sleep(short_sleep_time)

print('获取fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')
send_dingding_msg('获取fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')
raise ValueError('获取fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')

def get_candle_data(exchange, symbol, time_interval, run_time, max_try_amount, candle_num):
"""
:param exchange:
:param symbol_config:
:param time_interval:
:param run_time:
:param max_try_amount:
:param symbol:
:param candle_num:
:return:
尝试获取K线数据，并检验质量
"""
# 标记开始时间
start_time = datetime.now()
print('start get_candle_data', symbol, start_time)

# 尝试获取数据
for i in range(max_try_amount):
      # 获取symbol该品种最新的K线数据
      df = ccxt_fetch_candle_data(exchange, symbol, time_interval, limit=candle_num)
      if df.empty:
         continue
      # 判断是否包含最新的数据
      _ = df[df['candle_begin_time_GMT8'] == (run_time - timedelta(minutes=int(time_interval.strip('m'))))]
      if _.empty:
         print('获取数据不包含最新的数据，重新获取')
         time.sleep(short_sleep_time)
         continue

      else:
         # 获取到了最新数据

         df = df[df['candle_begin_time_GMT8'] < pd.to_datetime(run_time)]  # 去除run_time周期的数据
         print('end get_candle_data', symbol, datetime.now() - start_time)
         return  df

print('获取candle_data数据次数超过max_try_amount，数据返回空值')
return pd.DataFrame()

def next_run_time(time_interval, ahead_time=1):
if time_interval.endswith('m'):
      now_time = datetime.now()
      time_interval = int(time_interval.strip('m'))
      target_min = (int(now_time.minute / time_interval) + 1) * time_interval
      if target_min < 60:
         target_time = now_time.replace(minute=target_min, second=0, microsecond=0)
      else:
         if now_time.hour == 23:
            target_time = now_time.replace(hour=0, minute=0, second=0, microsecond=0)
            target_time += timedelta(days=1)
         else:
            target_time = now_time.replace(hour=now_time.hour + 1, minute=0, second=0, microsecond=0)
      # sleep直到靠近目标时间之前
      if (target_time - datetime.now()).seconds < ahead_time+1:
         print('距离target_time不足', ahead_time, '秒，下下个周期再运行')
         target_time += timedelta(minutes=time_interval)
      print('下次运行时间', target_time)
      return target_time
else:
      exit('time_interval doesn\'t end with m')

# =====主函数
def main():
# ===钉钉内容
max_len = 300
max_try_amount = 10
candle_num =3
config = pd.read_csv('swap_usdt.csv', index_col="coin")
strategy_name=strategy.upper()
msg_content = 'SWAP_%s_%s策略报告：' % (trade_coin , strategy_name)
closebyhand = config['closebyhand'][coin]  # 手贱模式
leverage = config["leverage"][coin]
fv = config['fv'][coin]
stop_lose_pct = int(config["stop_lose_pct"][coin])
client_oid = trade_coin + time_interval + str(closebyhand) + 'usdt' + strategy
para = [int(config[strategy][coin])]
account_info = fetch_account_info(exchange, symbol, client_oid=client_oid)
msg_content += '\n持有' + str(base_coin) + ':' + str(round(account_info['账户余额'], 2))

if account_info['多头仓位数量'] != 0:
      msg_content += '\n多头仓位数量:' + str(account_info['多头仓位数量'])
      msg_content += '\n仓位成本价:' + str(account_info['仓位成本'])
elif account_info['空头仓位数量'] != 0:
      msg_content += '\n空头仓位数量:' + str(account_info['空头仓位数量'])
      msg_content += '\n仓位成本价:' + str(account_info['仓位成本'])
else:
      msg_content += '\n无持仓信息'

history_candle_data = get_history_candle_data(exchange, symbol,time_interval, max_len= max_len)

run_time = next_run_time(time_interval)
sleep(max(0, (run_time - datetime.now()).seconds))
while True:  # 在靠近目标时间时
      if datetime.now() < run_time:
         continue
      else:
         break

recent_candle_data=get_candle_data(exchange, symbol, time_interval, run_time, max_try_amount, candle_num)

df = history_candle_data.append(recent_candle_data, ignore_index=True)

df.drop_duplicates(subset=['candle_begin_time_GMT8'], keep='last', inplace=True)
df.sort_values(by='candle_begin_time_GMT8', inplace=True)  # 排序，理论上这步应该可以省略，加快速度
df = df.iloc[-max_len:]  # 保持最大K线数量不会超过max_len个
df.reset_index(drop=True, inplace=True)

new_price = df.iloc[-1]['close']
msg_content += '\n' + str(trade_coin) + '实时价格:' + str(new_price)

df = getattr(Signal,strategy)(df, para=para)
signal = df.iloc[-1]['signal']
msg_content += '\n策略信号:' + str(signal)

# ===判断交易方向并且下单，除了无交易信号之外，总共有6种情况
if signal == -1:
      if account_info['多头仓位数量'] > 0:
         size = int(account_info['多头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平多',
                     price=new_price * 0.98,
                     match_price="0", size=size)
         # 更新账户信息、价格信息
         sleep(1)
         account_info = fetch_account_info(exchange, symbol, client_oid=client_oid)
         new_price = exchange.fetch_ticker(symbol)['ask']
         size = int(account_info['账户余额'] * leverage / (new_price * fv))
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='开空',
                     price=new_price * 0.98,
                     match_price="0", size=size)
         msg_content += '\n关闭多单并开空单'
         msg_content += '\n开空交易数量:' + str(size) + '\n'
      elif account_info["空头仓位数量"] > 0:
         msg_content += '\n之前已开空单，本次不开单'
         pass
      else:
         size = int(account_info['账户余额'] * leverage / (new_price * fv))
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='开空',
                     price=new_price * 0.98,
                     match_price="0", size=size)
         msg_content += '\n开空交易数量:' + str(size) + '\n'
elif signal == 1:
      if account_info["空头仓位数量"] > 0:
         size = int(account_info['空头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平空',
                     price=new_price * 1.02,
                     match_price="0", size=size)

         # 更新账户信息、价格信息
         sleep(1)
         account_info = fetch_account_info(exchange, symbol, client_oid=client_oid)
         new_price = exchange.fetch_ticker(symbol)['bid']
         size = int(account_info['账户余额'] * leverage / (new_price * fv))
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='开多',
                     price=new_price * 1.02,
                     match_price="0", size=size)
         msg_content += '\n关闭空单并开多单'
         msg_content += '\n开多交易数量:' + str(size) + '\n'
      elif account_info["多头仓位数量"] > 0:
         msg_content += '\n之前已开多单，本次不开单'
         pass
      else:
         size = int(account_info['账户余额'] * leverage / (new_price * fv))
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='开多',
                     price=new_price * 1.02,
                     match_price="0", size=size)
         msg_content += '\n开多交易数量:' + str(size) + '\n'
elif signal == 0:
      if account_info["空头仓位数量"] > 0:
         size = int(account_info['空头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平空',
                     price=new_price * 1.02,
                     match_price="0", size=size)
         msg_content += '\n止损空单交易数量:' + str(size) + '\n'
      elif account_info["多头仓位数量"] > 0:
         size = int(account_info['多头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平多',
                     price=new_price * 0.98,
                     match_price="0", size=size)
         msg_content += '\n止损多单交易数量:' + str(size) + '\n'
      else:
         msg_content += '\n之前已平仓，本次不开单'
         pass
elif account_info["多头仓位数量"] > 0:
      if df.iloc[-1]['close'] < float(account_info['仓位成本']) * (1 - stop_lose_pct / 100):
         size = int(account_info['多头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平多',
                     price=new_price * 0.98,
                     match_price="0", size=size)
         msg_content += '\n止损多单交易数量:' + str(size) + '\n'
elif account_info["空头仓位数量"] > 0:
      if df.iloc[-1]['close'] > float(account_info['仓位成本']) * (1 + stop_lose_pct / 100):
         size = int(account_info['空头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平空',
                     price=new_price * 1.02,
                     match_price="0", size=size)
         msg_content += '\n止损空单交易数量:' + str(size) + '\n'
else:
      msg_content += '\n本周期无操作'
      print('本周期无操作')

try:
      send_dingding_msg(msg_content)
except Exception as e:
      print(e)
# #  =====运行主体
while True:
try:
      main()
      time.sleep(10)
except Exception as e:
      send_dingding_msg('系统出错，10s之后重新运行')
      print(e)
      time.sleep(10)交割合约版本：
"""
# 程序主要思路如下：
通过while语句，每根K线不断的循环。
每次循环中需要做的操作步骤
1. 更新账户信息
2. 获取实时数据
3. 根据最新数据计算买卖信号
4. 根据目前仓位、买卖信息，结束本次循环，或者进行交易
5. 交易
"""
from urllib.request import urlopen
import ccxt
import time
import pandas as pd
from datetime import datetime, timedelta
from time import sleep
import json
import warnings
import requests
import random
import hashlib
import hmac
import base64
from urllib import parse
import numpy as np
import Signal
warnings.filterwarnings("ignore")
pd.set_option('expand_frame_repr', False)  # 当列太多时不换行
pd.set_option("display.max_columns",100)

coin="BTC-USDT-15"
strategy_name ='boll'

config =pd.read_csv('futures_usdt.csv',index_col="coin")
exchange = ccxt.okex3()
exchange.load_markets()
exchange.apiKey = config["apiKey"][coin]
exchange.secret = config["secret"][coin]
exchange.password=config["password"][coin]
symbol = config["instrument_id"][coin]

secret = config["robot_secret"][coin]
robot_id = config["robot_id"][coin]
def send_dingding_msg(content, secret=secret, robot_id=robot_id):
timestamp = int(round(time.time() * 1000))
secret_enc = bytes(secret.encode('utf-8'))
string_to_sign = '{}\n{}'.format(timestamp, secret)
string_to_sign_enc = bytes(string_to_sign.encode('utf-8'))
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = parse.quote_plus(base64.b64encode(hmac_code))
query_string = 'access_token=' + str(robot_id) + "&timestamp=" + str(timestamp) + "&sign=" + str(sign)
try:
      msg = {
         "msgtype": "text",
         "text": {"content": content + "\n" + datetime.now().strftime("%Y-%m-%d %H:%M:%S")}}
      url = 'https://oapi.dingtalk.com/robot/send'
      url = url + '?' + query_string
      body = json.dumps(msg)
      # 设置重连次数
      requests.adapters.DEFAULT_RETRIES = 15
      # 设置连接活跃状态为False
      s = requests.session()
      s.keep_alive = False
      #  进行请求
      requests.post(url, data=body, headers={"Content-Type": "application/json ;charset=utf-8"}, timeout=5)
      print('成功发送钉钉')
except Exception as e:
      print('钉钉发送失败,尝试重复发送', e)
      '''请求超时'''
      for i in range(1, 5):
         print('请求超时，第%s次重复请求' % i)
         response = requests.post(url, data=body, headers={"Content-Type": "application/json ;charset=utf-8"},
                                 timeout=5)
         requests.adapters.DEFAULT_RETRIES = 5
         if response.status_code == 200:
            print('成功发送钉钉')
            return response

def place_order(exchange, symbol, client_oid, order_type, buy_or_sell, price, match_price, size):
"""
下单
:param lever_rate:
:param exchange:
:param symbol:
:param contract_type: 合约类型: this_week:当周 next_week:下周 quarter:季度
:param order_type:
:param buy_or_sell: 1:开多 2:开空 3:平多 4:平空
:param price:
:param amount: 这里的amount是合约张数
:return:
"""
if order_type == '1':
      match_price = '1'
elif order_type == '0':
      match_price = '0'
if buy_or_sell == '开多':
      type = '1'
elif buy_or_sell == '开空':
      type = '2'
elif buy_or_sell == '平多':
      type = '3'
elif buy_or_sell == '平空':
      type = '4'
# 下单
for i in range(10):
      try:
         order_info = exchange.futures_post_order(params={'client_oid': client_oid,
                                                         'type': type,
                                                         'instrument_id': symbol,
                                                         'order_type': 0,
                                                         'price': price,
                                                         'size': size,
                                                         'match_price': match_price,
                                                         'leverage': 3
                                                         }
                                                   )
         print('okex 合约交易下单成功：', symbol, buy_or_sell, price, size)
         print('下单信息：', order_info, '\n')
         return order_info

      except Exception as e:
         print(e)
         if "error_code" in str(e) and "32019" in str(e):
            _df = exchange.futures_get_instruments_instrument_id_price_limit(params={'instrument_id': symbol})
            limit_high = float(_df["highest"])
            limit_low = float(_df["lowest"])
            print("限价数据：", limit_high, limit_low, "下单价格：", price)
            if buy_or_sell in ["开多", "平空"]:
                  price = limit_high * 0.995
            else:
                  price = limit_low * 1.005
            print("修改后的下单价格：", price)
            time.sleep(0.5)
         if "error_code" in str(e) and "32025" in str(e):
            time.sleep(30)
         if "error_code" in str(e) and "30014" in str(e):
            time.sleep(2)
raise ValueError('okex 合约交易下单报错次数过多，程序终止')
print('下单报错次数过多，程序终止')
send_dingding_msg('下单报错次数过多，程序终止')

def next_run_time(time_interval, ahead_time=1):
if time_interval.endswith('m'):
      now_time = datetime.now()
      time_interval = int(time_interval.strip('m'))
      target_min = (int(now_time.minute / time_interval) + 1) * time_interval
      if target_min < 60:
         target_time = now_time.replace(minute=target_min, second=0, microsecond=0)
      else:
         if now_time.hour == 23:
            target_time = now_time.replace(hour=0, minute=0, second=0, microsecond=0)
            target_time += timedelta(days=1)
         else:
            target_time = now_time.replace(hour=now_time.hour + 1, minute=0, second=0, microsecond=0)
      # sleep直到靠近目标时间之前
      if (target_time - datetime.now()).seconds < ahead_time + 1:
         print('距离target_time不足', ahead_time, '秒，下下个周期再运行')
         target_time += timedelta(minutes=time_interval)
      print('下次运行时间', target_time)
      return target_time
else:
      exit('time_interval doesn\'t end with m')

# ===获取账户和仓位信息
def fetch_account_info(exchange, trade_symbol, client_oid):
"""
获取账户信息
:param exchange: 交易所
:param base_coin: 基准币的名称
:param trade_symbol: 交易对的名称
:return:
:param account_info: dict形式，包含账户和仓位信息
"""
# 创建用于存放账户信息的变量
account_info = {'账户余额': 0, '多头仓位数量': 0, '空头仓位数量': 0, '仓位成本': 0}
while True:
      # 获取账户的资产信息
      try:
         data = exchange.futures_get_accounts_currency(params={'currency': trade_symbol})  # 从交易所获取账户balance信息
      except Exception as e:
         send_dingding_msg('获取账户信息失败')
         print(e)
         sleep(0.5)
         continue
      account_info['账户余额'] = float(data["equity"])
      # 获取账户的order

      try:
         client_oid_info = exchange.futures_get_orders_instrument_id_client_oid(
            params={'instrument_id': symbol, "client_oid": client_oid})

      except Exception as e:
         send_dingding_msg('获取信息失败')
         print(e)
         sleep(0.5)
         continue
      if client_oid_info:
         if client_oid_info['state'] == "2":
            if client_oid_info['type'] == "1":  # 当持仓信息信息不为空时
                  account_info['多头仓位数量'] = float(client_oid_info['filled_qty'])
                  account_info['仓位成本'] = float(client_oid_info['price_avg'])
            elif client_oid_info['type'] == "2":  # 当持仓信息信息不为空时
                  account_info['空头仓位数量'] = float(client_oid_info['filled_qty'])
                  account_info['仓位成本'] = float(client_oid_info['price_avg'])
            elif client_oid_info['type'] == "3" or "4":  # 当持仓信息信息不为空时
                  account_info['空头仓位数量'] = 0
                  account_info['多头仓位数量'] = 0
                  account_info['仓位成本'] = 0
      break
return account_info

def get_history_candle_data(exchange, symbol, time_interval, max_len, max_try_amount=20):
"""
获取某个币种在okex交易所所有能获取的历史数据，目前v3接口最多获取1440根
:param exchange:
:param symbol:
:param time_interval_str:
:param now_milliseconds:
:param max_len:
:param max_try_amount:
:return:
"""
now_milliseconds = int(time.time() * 1e3)  # 获取统一的数据截止时间
time_interval_int = int(time_interval[:-1])
time_segment = time_interval_int * 60 * 1000
end = now_milliseconds - time_segment
since = end - max_len * time_segment
all_kline_data = []
while end - since >= time_segment:
      kline_data = []
      for i in range(max_try_amount):
         try:
            kline_data = exchange.fetch_ohlcv(symbol=symbol, since=since, timeframe=time_interval)
            break
         except Exception as e:
            print(e)
            try:
                  e_str = str(e)
                  if '35004' in e_str:
                     time.sleep(30)
                  if '30014' in e_str:
                     time.sleep(10)
            except:
                  time.sleep(10)

            if i == (max_try_amount - 1):
                  print('获取历史数据时，fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')
                  send_dingding_msg('获取历史数据时，fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')
                  raise ValueError('获取历史数据时，fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')
      if kline_data:
         # 更新获取时间
         since = kline_data[-1][0]
         all_kline_data += kline_data
      else:
         break

# 对数据进行整理
df = pd.DataFrame(all_kline_data, dtype=float)
df.rename(columns={0: 'MTS', 1: 'open', 2: 'high', 3: 'low', 4: 'close', 5: 'volume'}, inplace=True)
df['candle_begin_time'] = pd.to_datetime(df['MTS'], unit='ms')
df['candle_begin_time_GMT8'] = df['candle_begin_time'] + timedelta(hours=8)
df = df[['candle_begin_time_GMT8', 'open', 'high', 'low', 'close', 'volume']]

# 删除重复的数据
df.drop_duplicates(subset=['candle_begin_time_GMT8'], keep='last', inplace=True)
df.reset_index(drop=True, inplace=True)
df = df[:-1]

return df

def ccxt_fetch_candle_data(exchange, symbol, time_interval, limit, max_try_amount=5):
"""
:param exchange:
:param symbol:
:param time_interval:
:param limit:
:param max_try_amount:
:return:
获取K线数据
"""
for _ in range(max_try_amount):
      try:
         # 获取数据
         data = exchange.fetch_ohlcv(symbol=symbol, timeframe=time_interval, limit=limit)
         # 整理数据
         df = pd.DataFrame(data, dtype=float)
         df.rename(columns={0: 'MTS', 1: 'open', 2: 'high',
                           3: 'low', 4: 'close', 5: 'volume'}, inplace=True)
         df['candle_begin_time'] = pd.to_datetime(df['MTS'], unit='ms')
         df['candle_begin_time_GMT8'] = df['candle_begin_time'] + timedelta(hours=8)
         df = df[['candle_begin_time_GMT8', 'open', 'high', 'low', 'close', 'volume']]
         return df
      except Exception as e:
         print(e)
         print('获取fetch_ohlcv合约K线数据失败，稍后重试')
         time.sleep(short_sleep_time)

print('获取fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')
send_dingding_msg('获取fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')
raise ValueError('获取fetch_ohlcv合约K线数据失败，失败次数过多，程序退出')

def get_candle_data(exchange, symbol, time_interval, run_time, max_try_amount, candle_num):
"""
:param exchange:
:param symbol_config:
:param time_interval:
:param run_time:
:param max_try_amount:
:param symbol:
:param candle_num:
:return:
尝试获取K线数据，并检验质量
"""
# 标记开始时间
start_time = datetime.now()
print('start get_candle_data', symbol, start_time)

# 尝试获取数据
for i in range(max_try_amount):
      # 获取symbol该品种最新的K线数据
      df = ccxt_fetch_candle_data(exchange, symbol, time_interval, limit=candle_num)
      if df.empty:
         continue
      # 判断是否包含最新的数据
      _ = df[df['candle_begin_time_GMT8'] == (run_time - timedelta(minutes=int(time_interval.strip('m'))))]
      if _.empty:
         print('获取数据不包含最新的数据，重新获取')
         time.sleep(short_sleep_time)
         continue

      else:
         # 获取到了最新数据

         df = df[df['candle_begin_time_GMT8'] < pd.to_datetime(run_time)]  # 去除run_time周期的数据
         print('end get_candle_data', symbol, datetime.now() - start_time)
         return  df

print('获取candle_data数据次数超过max_try_amount，数据返回空值')
return pd.DataFrame()

def main():
max_len = 1400
max_try_amount = 10
candle_num = 5
config = pd.read_csv('futures_usdt.csv', index_col="coin")
time_interval = config["time_interval"][coin]
trade_coin = symbol.split("-")[0]
trade_symbol = symbol[0:8].lower()
base_coin = symbol.split("-")[1]
msg_content = 'Fr_%s_%s策略%s报告：' % (trade_coin, strategy_name.upper(),time_interval)

closebyhand = config['closebyhand'][coin]  # 手贱模式
leverage = config["leverage"][coin]
fv = config['fv'][coin]
stop_lose_pct = int(config["stop_lose_pct"][coin])

client_oid = trade_coin + time_interval + str(closebyhand) + strategy_name
if strategy_name =='boll':
      para = [int(config["boll_1"][coin]), float(config['boll_2'][coin])]
else:
      para = [int(config[strategy_name][coin])]

account_info = fetch_account_info(exchange, trade_symbol, client_oid=client_oid)
msg_content += '\n持有' + str(base_coin) + ':' + str(round(account_info['账户余额'], 2))

if account_info['多头仓位数量'] != 0:
      msg_content += '\n多头仓位数量:' + str(account_info['多头仓位数量'])
      msg_content += '\n仓位成本价:' + str(account_info['仓位成本'])
elif account_info['空头仓位数量'] != 0:
      msg_content += '\n空头仓位数量:' + str(account_info['空头仓位数量'])
      msg_content += '\n仓位成本价:' + str(account_info['仓位成本'])
else:
      msg_content += '\n无持仓信息'

history_candle_data = get_history_candle_data(exchange, symbol, time_interval, max_len=max_len)

run_time = next_run_time(time_interval)
sleep(max(0, (run_time - datetime.now()).seconds))
while True:  # 在靠近目标时间时
      if datetime.now() < run_time:
         continue
      else:
         break

recent_candle_data = get_candle_data(exchange, symbol, time_interval, run_time, max_try_amount, candle_num)
df = history_candle_data.append(recent_candle_data, ignore_index=True)
df.drop_duplicates(subset=['candle_begin_time_GMT8'], keep='last', inplace=True)
df = df.iloc[-max_len:]  # 保持最大K线数量不会超过max_len个
df.reset_index(drop=True, inplace=True)
new_price = df.iloc[-1]['close']
msg_content += '\n' + str(trade_coin) + '实时价格:' + str(new_price)

df = getattr(Signal, strategy_name)(df, para=para)
signal = df.iloc[-1]['signal']

msg_content += '\n策略信号:' + str(signal)

   # ===判断交易方向并且下单，除了无交易信号之外，总共有6种情况
if signal == -1:
      if account_info['多头仓位数量'] > 0:
         size = int(account_info['多头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平多',
                     price=new_price * 0.98,
                     match_price="0", size=size)
         # 更新账户信息、价格信息
         sleep(1)
         account_info = fetch_account_info(exchange, trade_symbol, client_oid=client_oid)
         new_price = exchange.fetch_ticker(symbol)['ask']
         size = int(account_info['账户余额'] * leverage / (new_price * fv))
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='开空',
                     price=new_price * 0.98,
                     match_price="0", size=size)
         msg_content += '\n关闭多单并开空单'
         msg_content += '\n开空交易数量:' + str(size) + '\n'
      elif account_info["空头仓位数量"] > 0:
         msg_content += '\n之前已开空单，本次不开单'
         pass
      else:
         size = int(account_info['账户余额'] * leverage / (new_price * fv))
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='开空',
                     price=new_price * 0.98,
                     match_price="0", size=size)
         msg_content += '\n开空交易数量:' + str(size) + '\n'
elif signal == 1:
      if account_info["空头仓位数量"] > 0:
         size = int(account_info['空头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平空',
                     price=new_price * 1.02,
                     match_price="0", size=size)
         # 更新账户信息、价格信息
         sleep(1)
         account_info = fetch_account_info(exchange, trade_symbol, client_oid=client_oid)
         new_price = exchange.fetch_ticker(symbol)['bid']
         size = int(account_info['账户余额'] * leverage / (new_price * fv))
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='开多',
                     price=new_price * 1.02,
                     match_price="0", size=size)
         msg_content += '\n关闭空单并开多单'
         msg_content += '\n开多交易数量:' + str(size) + '\n'
      elif account_info["多头仓位数量"] > 0:
         msg_content += '\n之前已开多单，本次不开单'
         pass
      else:
         size = int(account_info['账户余额'] * leverage / (new_price * fv))
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='开多',
                     price=new_price * 1.02,
                     match_price="0", size=size)
         msg_content += '\n开多交易数量:' + str(size) + '\n'
elif signal == 0:
      if account_info["空头仓位数量"] > 0:
         size = int(account_info['空头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平空',
                     price=new_price * 1.02,
                     match_price="0", size=size)
         msg_content += '\n止损空单交易数量:' + str(size) + '\n'
      elif account_info["多头仓位数量"] > 0:
         size = int(account_info['多头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平多',
                     price=new_price * 0.98,
                     match_price="0", size=size)
         msg_content += '\n止损多单交易数量:' + str(size) + '\n'
      else:
         msg_content += '\n之前已平仓，本次不开单'
         pass
elif account_info["多头仓位数量"] > 0:
      if df.iloc[-1]['close'] < float(account_info['仓位成本']) * (1 - stop_lose_pct / 100):
         size = int(account_info['多头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平多',
                     price=new_price * 0.98,
                     match_price="0", size=size)
         msg_content += '\n止损多单交易数量:' + str(size) + '\n'
elif account_info["空头仓位数量"] > 0:
      if df.iloc[-1]['close'] > float(account_info['仓位成本']) * (1 + stop_lose_pct / 100):
         size = int(account_info['空头仓位数量'])
         place_order(exchange=exchange, client_oid=client_oid, symbol=symbol, order_type='0', buy_or_sell='平空',
                     price=new_price * 1.02,
                     match_price="0", size=size)
         msg_content += '\n止损空单交易数量:' + str(size) + '\n'
else:
      msg_content += '\n本周期无操作'
      print('本周期无操作')

try:
      send_dingding_msg(msg_content)
except Exception as e:
      print(e)

sleep(10 * 1)
# #  =====运行主体
while True:
try:
      main()
      time.sleep(10)
except Exception as e:
      send_dingding_msg('系统出错，10s之后重新运行')
      print(e, type(e))
      time.sleep(10)csv格式：
交割合约：

永续合约：

上架时就把
“
coin="BTC-USDT-15"
strategy_name ='boll'
”
这个部分修改就行
然后在Linux服务器上用nohup命令：
nohup python -u btc_vix.py >1btc_vix.log 2>&1 &
搞定。这个完全是我实盘框架。比较粗糙，里面的函数可封装、模块化。框架可优化、简洁的地方比较多。
另外就差一个Signal文件啦，这个就由各位自行搞定啦。

【python数字货币量化_21精华帖_策略第9篇】永续合约、交割合约client_oid版本代码迭代1.0

浏览过的版块