初探 Python 量化交易

搭建交易执行层核心

import base64
import datetime
import hashlib
import hmac
import json
import time

import requests

base_url = "https://api.sandbox.gemini.com"
endpoint = "/v1/order/new"
url = base_url + endpoint

gemini_api_key = "account-zmidXEwP72yLSSybXVvn"
gemini_api_secret = "375b97HfE7E4tL8YaP3SJ239Pky9".encode()

t = datetime.datetime.now()
payload_nonce = str(int(time.mktime(t.timetuple()) * 1000))

payload = {
    "request": "/v1/order/new",
    "nonce": payload_nonce,
    "symbol": "btcusd",
    "amount": "5",
    "price": "3633.00",
    "side": "buy",
    "type": "exchange limit",
    "options": ["maker-or-cancel"]
}

encoded_payload = json.dumps(payload).encode()
b64 = base64.b64encode(encoded_payload)
signature = hmac.new(gemini_api_secret, b64, hashlib.sha384).hexdigest()

request_headers = {
    'Content-Type': "text/plain",
    'Content-Length': "0",
    'X-GEMINI-APIKEY': gemini_api_key,
    'X-GEMINI-PAYLOAD': b64,
    'X-GEMINI-SIGNATURE': signature,
    'Cache-Control': "no-cache"
}

response = requests.post(url,
                         data=None,
                         headers=request_headers)

new_order = response.json()
print(new_order)

行情数据对接和抓取

import copy
import json
import ssl
import time

import websocket


class OrderBook(object):
    BIDS = 'bid'
    ASKS = 'ask'

    def __init__(self, limit=20):
        self.limit = limit

        # (price, amount)
        self.bids = {}
        self.asks = {}

        self.bids_sorted = []
        self.asks_sorted = []

    def insert(self, price, amount, direction):
        if direction == self.BIDS:
            if amount == 0:
                if price in self.bids:
                    del self.bids[price]
            else:
                self.bids[price] = amount
        elif direction == self.ASKS:
            if amount == 0:
                if price in self.asks:
                    del self.asks[price]
            else:
                self.asks[price] = amount
        else:
            print('WARNING: unknown direction {}'.format(direction))

    def sort_and_truncate(self):
        # sort
        self.bids_sorted = sorted([(price, amount) for price, amount in self.bids.items()], reverse=True)
        self.asks_sorted = sorted([(price, amount) for price, amount in self.asks.items()])

        # truncate
        self.bids_sorted = self.bids_sorted[:self.limit]
        self.asks_sorted = self.asks_sorted[:self.limit]

        # copy back to bids and asks
        self.bids = dict(self.bids_sorted)
        self.asks = dict(self.asks_sorted)

    def get_copy_of_bids_and_asks(self):
        return copy.deepcopy(self.bids_sorted), copy.deepcopy(self.asks_sorted)


class Crawler:
    def __init__(self, symbol, output_file):
        self.order_book = OrderBook(limit=10)
        self.output_file = output_file

        self.ws = websocket.WebSocketApp('wss://api.gemini.com/v1/marketdata/{}'.format(symbol),
                                         on_message=lambda ws, message: self.on_message(message))
        self.ws.run_forever(sslopt={'cert_reqs': ssl.CERT_NONE})

    def on_message(self, message):
        data = json.loads(message)
        for event in data['events']:
            price, amount, direction = float(event['price']), float(event['remaining']), event['side']
            self.order_book.insert(price, amount, direction)

        self.order_book.sort_and_truncate()

        with open(self.output_file, 'a+') as f:
            bids, asks = self.order_book.get_copy_of_bids_and_asks()
            output = {
                'bids': bids,
                'asks': asks,
                'ts': int(time.time() * 1000)
            }
            f.write(json.dumps(output) + '\n')


if __name__ == '__main__':
    crawler = Crawler(symbol='BTCUSD', output_file='BTCUSD.txt')

策略与回测系统

Utils.py

import os.path as path

import pandas as pd


def assert_msg(condition, msg):
    if not condition:
        raise Exception(msg)


def read_file(filename):
    # 获得文件绝对路径
    filepath = path.join(path.dirname(__file__), filename)
    assert_msg(path.exists(filepath), "文件不存在")
    return pd.read_csv(filepath,
                       index_col=0,
                       parse_dates=True,
                       infer_datetime_format=True)


def SMA(values, n):
    """
    Simple Moving Average 简单移动均值
    """
    return pd.Series(values).rolling(n).mean()


def crossover(series_1, series_2) -> bool:
    """
    这俩指标，一个是小窗口的 SMA，一个是大窗口的 SMA；
    如果小窗口的 SMA 从下面刺破或者穿过大窗口 SMA，那么说明，投资品的价格在短期内快速上涨，同时这个趋势很强烈，可能是一个买入的信号；
    反之，如果大窗口的 SMA 从下方突破小窗口 SMA，那么说明，投资品的价格在短期内快速下跌，我们应该考虑卖出。
    """
    return series_1[-2] < series_2[-2] and series_1[-1] > series_2[-1]

Strategy.py

import abc
from typing import Callable

import numpy as np


class Strategy(metaclass=abc.ABCMeta):
    def __init__(self, broker, data):
        self._broker = broker
        self._data = data
        self._tick = 0

    def I(self, func: Callable, *args) -> np.ndarray:
        """
        计算买卖指标向量，用于判定这个时间点上需要进行"买"还是"卖"
        """
        value = func(*args)
        value = np.asarray(value)
        assert_msg(value.shape[-1] == len(self._data.Close), "指标长度必须和 data 长度相同")
        return value

    @property
    def tick(self):
        return self._tick

    @abc.abstractmethod
    def init(self):
        pass

    @abc.abstractmethod
    def next(self, tick):
        """
        步进函数，执行第 tick 步的策略；tick 代表当前的"时间"
        """
        pass

    def buy(self):
        self._broker.buy()

    def sell(self):
        self._broker.sell()

    @property
    def data(self):
        return self._data


from Utils import assert_msg, crossover, SMA


class SmaCross(Strategy):
    # 小窗口 SMA 的窗口大小，用于计算 SMA 快线
    fast = 30

    # 大窗口 SMA 的窗口大小，用于计算 SMA 慢线
    slow = 90

    def init(self):
        # 计算历史上每个时刻的快线和慢线
        self.sma1 = self.I(SMA, self.data.Close, self.fast)
        self.sma2 = self.I(SMA, self.data.Close, self.slow)

    def next(self, tick):
        # 如果快线刚好越过慢线，买入全部
        if crossover(self.sma1[:tick], self.sma2[:tick]):
            self.buy()
        # 如果慢线刚好越过快线，卖出全部
        elif crossover(self.sma2[:tick], self.sma1[:tick]):
            self.sell()
        # 否则，不执行任何操作
        else:
            pass

Backtest.py

from numbers import Number

import numpy as np
import pandas as pd

from Strategy import Strategy, SmaCross
from Utils import read_file, assert_msg


class ExchangeAPI:
    def __init__(self, data, cash, commission):
        assert_msg(0 < cash, "初始现金数量必须大于 0，输入的现金数量：{}".format(cash))
        assert_msg(0 <= commission <= 0.05, "合理的手续费率一般不会超过 5%，输入的费率：{}".format(commission))
        self._inital_cash = cash
        self._data = data
        self._commission = commission
        self._position = 0
        self._cash = cash
        self._i = 0

    @property
    def cash(self):
        """
        :return: 返回当前账户现金数量
        """
        return self._cash

    @property
    def position(self):
        """
        :return: 返回当前账户仓位
        """
        return self._position

    @property
    def initial_cash(self):
        return self._inital_cash

    @property
    def market_value(self):
        return self._cash + self._position * self.current_price

    @property
    def current_price(self):
        return self._data.Close[self._i]

    def buy(self):
        self._position = float(self._cash * (1 - self._commission) / self.current_price)
        self._cash = 0.0

    def sell(self):
        self._cash += float(self._position * self.current_price * (1 - self._commission))
        self._position = 0.0

    def next(self, tick):
        self._i = tick


class Backtest:
    def __init__(self,
                 data: pd.DataFrame,
                 strategy_type: type(Strategy),
                 broker_type: type(ExchangeAPI),
                 cash: float = 10000,
                 commission: float = .0):
        assert_msg(issubclass(strategy_type, Strategy), "strategy_type 不是 Strategy 类型")
        assert_msg(issubclass(broker_type, ExchangeAPI), "broker_type 不是 ExchangeAPI 类型")
        assert_msg(isinstance(commission, Number), "commission 不是浮点数值类型")

        data = data.copy(False)

        if 'Volume' not in data:
            data['Volume'] = np.nan

        # 检查缺失值
        assert_msg(not data[['Open', 'High', 'Low', 'Close']].max().isnull().any(),
                   ("部分 OHLC 包含缺失值，请去掉那些行或者通过差值填充"))

        # 如果行情数据没有按照时间排序，进行重新排序
        if not data.index.is_monotonic_increasing:
            data = data.sort_index()

        self._data = data
        self._broker = broker_type(data, cash, commission)
        self._strategy = strategy_type(self._broker, self._data)
        self._results = None

    def run(self) -> pd.Series:
        strategy = self._strategy
        broker = self._broker

        strategy.init()

       # 设定回测开始位置和结束位置
        start = 100
        end = len(self._data)

        # 更新市场状态，然后再执行策略
        for i in range(start, end):
            # 注意要先把市场状态移动到第 i 个时刻
            broker.next(i)
            strategy.next(i)

        self._results = self._compute_result(broker)
        return self._results

    def _compute_result(self, broker):
        s = pd.Series()
        s['初始市值'] = broker.initial_cash
        s['结束市值'] = broker.market_value
        s['收益'] = broker.market_value - broker.initial_cash
        return s


def main():
    BTCUSD = read_file('BTCUSD_GEMINI.csv')
    ret = Backtest(BTCUSD, SmaCross, ExchangeAPI, 10000.0, 0.003).run()
    print(ret)


if __name__ == '__main__':
    main()