量化學(xué)習(xí)中的一點(diǎn)編程體會

在我的 19 年的年度學(xué)習(xí)計劃中续搀，量化交易是其中的一項(xiàng)大的工程焰薄，是自己下定決心要攻克的一個難題。應(yīng)該說 19 年是我的量化學(xué)習(xí)年止喷，為了強(qiáng)化自己的決定，我付費(fèi)在年初參加了邢不行的量化學(xué)習(xí)課程混聊，到現(xiàn)在已經(jīng)過了一個多月弹谁，自己從沒有摸過 python , 到能用 python 編寫簡單的量化程序。除了認(rèn)真學(xué)習(xí)課程外，也下了很大的功夫和精力预愤。

我的第一個布林策略也已經(jīng)運(yùn)行了一段時間了沟于，程序的改進(jìn)也一直在進(jìn)行。從測試的情況看植康，效果應(yīng)該還不錯旷太。

在改進(jìn)程序的過程中，我對最初的量化程序進(jìn)行了重構(gòu)销睁，盡量將量化策略和相關(guān)的交易所代碼分離供璧。這樣一個量化策略寫完，只要適配一下交易所的代碼冻记，就可以在另外的交易所上運(yùn)行了睡毒。

在編程的過程中，對于一些常見問題的解決方法冗栗，有了一些體會演顾。我把它寫下來。

程序運(yùn)行的參數(shù)全部使用配置文件隅居，這樣運(yùn)行程序時钠至，使用不同的配置文件就可以了。

oscardeMacBook-Pro:oscbot oscar$ ./okex_boll.py 
Usage: okex_boll.py -c 策略配置文件 
okex 布林線策略
okex_boll.py: error: 請指定策略配置文件军浆！

我的主程序代碼：

#!/usr/bin/env python
# -*-coding:utf-8-*-

from bots.adaptation.trend_okex import TrendOkex
from bots.exchanges.okex_level import OkexLevel
from bots.algo.boll import BollAlgo
from bots.util.log import config_log
from bots.util.util import check_network, get_config
import logging

from optparse import OptionParser

parser = OptionParser(usage="%prog -c 策略配置文件 \nokex 布林線策略\n")

parser.add_option("-c", "--configfile",
                  action="store",
                  type='string',
                  dest="configfile",
                  help="指定策略運(yùn)行配置文件"
                  )

(options, args) = parser.parse_args()

if not options.configfile:
    parser.error('請指定策略配置文件棕洋！')

options = get_config(options.configfile)

config_log(options['name'], options['log'])

if not check_network():
    print('網(wǎng)絡(luò)錯誤,無法連接 google.com 請先檢查網(wǎng)絡(luò)或設(shè)置代理')
    exit(1)

logging.info('use config:%s', options['algo'])

ok = OkexLevel(options['algo']['trade'], options['algo']['base'], auths=options['api'])

tokex = TrendOkex(ok)

boll_algo = BollAlgo(tokex, para=[options['algo']['n'], options['algo']['m']], name=options['name'],
                     interval=options['algo']['interval'])

boll_algo.run()

我把所有策略的配置文件放在程序的 configs 目錄下，運(yùn)行產(chǎn)生的 log ,放在 logs 目錄下乒融。

配置文件示例：

{
  "name": "okex_boll",
  "log": "./logs/okex_boll.log",
  "algo": {
    "trade": "eth",
    "base": "usdt",
    "interval": 30,
    "n": 200,
    "m": 2
  },
  "api": {
    "name": "okex",
    "key": "**************",
    "secret": "******************",
    "pass": "***********"
  }
}

上面的布林參數(shù)我已經(jīng)改過了掰盘，請不要照抄使用！

程序運(yùn)行的 log 是非常重要的赞季，在程序有問題的時候愧捕，我們需要它來查找 bug.

以下是配置 log 的代碼

def config_log(name='oscbot',logfile=''):
    """
    設(shè)置 log 
    :param name:
    :return:
    """
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)  # Log等級總開關(guān)

    logger.name = name

    formatter = logging.Formatter("%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s")

    # log to console
    console = logging.StreamHandler()
    console.setLevel(logging.INFO)
    console.setFormatter(formatter)
    logger.addHandler(console)

    # log to file
    if logfile == '':
        logfile = './' + name + '.log'
    fh = logging.FileHandler(logfile, mode='a')
    fh.setLevel(logging.INFO)  # 輸出到file的log等級的開關(guān)
    fh.setFormatter(formatter)
    logger.addHandler(fh)

    # log to 釘釘
    config = json_config('config.json')
    dd = DingDing(robot_id=config['dingding']['robot_id'])
    formatter = logging.Formatter('%(name)s - %(message)s')
    dd.setFormatter(formatter)
    logger.addHandler(dd)

在寫交易所的代碼過程中，下單取帳戶數(shù)據(jù)要考慮失敗重試申钩，我使用 tenacity 庫使得這部分的代碼不會太繁瑣次绘。

from tenacity import retry, wait_exponential, stop_after_attempt
    @retry(wait=wait_exponential(multiplier=1, max=10), reraise=True, stop=stop_after_attempt(6))
    def get_kicker(self):
        try:
            return self.spotAPI.get_specific_ticker(self.instrument_id)
        except Exception as e:
            logging.error('取末成交訂單失敗:%s', e, extra={'dingding': True})
            logging.error(e, exc_info=True)
            raise e

需要了解的話，可以搜索一下它的用法撒遣。

為了加快速度邮偎，很多情況下從交易所取得的數(shù)據(jù)，可以緩存下來义黎，供以后使用禾进。所以我在代碼中使用了 lru_cache

from functools import lru_cache

@lru_cache(maxsize=32)
@retry(wait=wait_exponential(multiplier=1, max=10), reraise=True, stop=stop_after_attempt(6))
def level_config_info(api,instrument_id):
    try:
        return api.get_specific_config_info(instrument_id)
    except Exception as e:
        logging.warning('取杠桿配置信息報錯:%s', e, extra={'dingding': True})
        logging.error(e, exc_info=True)
        raise e

# 杠桿配置信息
def level_info(self,refresh=False):
        if refresh:
            level_config_info.cache_clear()
        return level_config_info(self.levelAPI,self.instrument_id)