手把手教你python实现量价形态选股知乎_【手把手教你】Python实现基于事件驱动的量化回测...
01引言
使用矢量化方法(pandas)建立的基于研究的量化回測框架，不考慮交易的委托成交行為，與真實市場情況差距比較大。今天為大家介紹的是基于事件驅動的回測框架，這是一種十分復雜的回測系統，力圖模擬實盤交易，搭建一種仿真的回測環境。
與矢量化方法相比，事件驅動的系統具有許多優點，一是事件驅動回測可以用于歷史回測和實時交易，而矢量化的回測必須一次獲得所有數據才能進行統計分析；二是使用事件驅動的回測不會出現前瞻性偏見，因為將市場數據接收視為“事件”，可以用市場數據“滴灌”來復制訂單管理和投資組合系統的行為方式；三是事件驅動的回測允許對如何執行訂單和產生交易成本進行定制。由于可以構建自定義交易處理程序，因此可以輕松處理基本的市場和限價訂單。
盡管事件驅動的回測系統具有許多優點，但與簡單的矢量化系統相比，兩大缺點也比較突出：一是實施和測試要復雜得多，有更多的“活動組建”(模塊)，導致引入錯誤的機會更大；二是執行速度較慢，進行數學計算時，無法利用最佳的矢量化運算。下面仍然與均值回歸交易策略為例，為大家展示Python基于事件驅動回測框架的構建思路，回測代碼主要參考了《Mastering Python for Finance》Chapter 9 Backtesting，對市場數據獲取使用了tushare作為替代。
02 回測框架與Python代碼
基于事件驅動的回測框架一般包括以下幾個模塊，(1)數據采集，數據采集模塊通過接口獲取行情數據和歷史數據(這里使用tushare)，產生市場數據事件。(2)事件模塊，一般是設定一個事件基類，然后在事件的基類下面生成很多子事件，如市場數據事件、交易信號事件、委托下單事件和訂單成交事件等。(3)策略模塊，一般先設定一個策略基類，然后通過基類衍生很多子策略，該模塊通過輸入數據，生成交易信號(signal)，即產生信號事件。(4)交易執行模塊，接收信號事件，確定需要開倉和平倉的頭寸數量，輸出委托下單事件，根據委托下單事件進行模擬或者真實的交易，當訂單成交事件完成時更新持有資產頭寸以及其他相關數據。(5)資產頭寸，記錄資金、倉位、倉位市值等信息。最后，所有事件通過事件隊列進行管理，當一個事件完成后，由下一個事件開始任務，不斷循環。
Python基于事件驅動的回測系統主要使用面向對象(class)來編寫，因此需要對類的基礎要求比較高。
#先引入后面可能用到的包(package)
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
%matplotlib inline
#正常顯示畫圖時出現的中文和負號
from pylab import mpl
mpl.rcParams['font.sans-serif']=['SimHei']
mpl.rcParams['axes.unicode_minus']=False
將每一個時間戳(timestamp)內的數據作為輸入參數，構建類TickData。
class TickData:
def __init__(self, symbol, timestamp,last_price=0, total_volume=0):
self.symbol = symbol
self.timestamp = timestamp
self.open_price = 0
self.last_price = last_price
self.total_volume = total_volume
根據數據要求，生成市場數據事件，這里主要獲取收盤價、開盤價、成交量和時間戳。
class MarketData:
def __init__(self):
self.__recent_ticks__ = dict()
def add_last_price(self, time, symbol, price, volume):
tick_data = TickData(symbol, time, price, volume)
self.__recent_ticks__[symbol] = tick_data
def add_open_price(self, time, symbol, price):
tick_data = self.get_existing_tick_data(symbol, time)
tick_data.open_price = price
def get_existing_tick_data(self, symbol, time):
if not symbol in self.__recent_ticks__:
tick_data = TickData(symbol, time)
self.__recent_ticks__[symbol] = tick_data
return self.__recent_ticks__[symbol]
def get_last_price(self, symbol):
return self.__recent_ticks__[symbol].last_price
def get_open_price(self, symbol):
return self.__recent_ticks__[symbol].open_price
def get_timestamp(self, symbol):
return self.__recent_ticks__[symbol].timestamp
獲取市場數據并搭建市場模擬交易的狀態。
#獲取數據
import tushare as ts
class MarketDataSource:
def __init__(self):
self.event_tick = None
self.ticker = None
self.autype='qfq'
self.start, self.end = None, None
self.md = MarketData()
def start_market_simulation(self):
data = ts.get_k_data(self.ticker,autype=self.autype,
start=self.start, end=self.end)
data.index=pd.to_datetime(data.date)
data=data.sort_index()
for time, row in data.iterrows():
self.md.add_last_price(time, self.ticker,
row["close"], row["volume"])
self.md.add_open_price(time, self.ticker, row["open"])
if not self.event_tick is None:
self.event_tick(self.md)
交易指令和頭寸管理。
class Order:
def __init__(self, timestamp, symbol, qty, is_buy,
is_market_order, price=0):
self.timestamp = timestamp
self.symbol = symbol
self.qty = qty
self.price = price
self.is_buy = is_buy
self.is_market_order = is_market_order
self.is_filled = False
self.filled_price = 0
self.filled_time = None
self.filled_qty = 0
class Position:
def __init__(self):
self.symbol = None
self.buys, self.sells, self.net = 0, 0, 0
self.realized_pnl = 0
self.unrealized_pnl = 0
self.position_value = 0
def event_fill(self, timestamp, is_buy, qty, price):
if is_buy:
self.buys += qty
else:
self.sells += qty
self.net = self.buys - self.sells
changed_value = qty * price * (-1 if is_buy else 1)
self.position_value += changed_value
if self.net == 0:
self.realized_pnl = self.position_value
def update_unrealized_pnl(self, price):
if self.net == 0:
self.unrealized_pnl = 0
else:
self.unrealized_pnl = price * self.net + self.position_value
return self.unrealized_pnl
策略的基類，其他策略都基于該策略進行編寫。
class Strategy:
def __init__(self):
self.event_sendorder = None
def event_tick(self, market_data):
pass
def event_order(self, order):
pass
def event_position(self, positions):
pass
def send_market_order(self, symbol, qty, is_buy, timestamp):
if not self.event_sendorder is None:
order = Order(timestamp, symbol, qty, is_buy, True)
self.event_sendorder(order)
下面以均值回歸模型為例進行回測，關于均值回歸模型更詳細的內容可參考推文：
class MeanRevertingStrategy(Strategy):
def __init__(self, symbol,
lookback_intervals=20,
buy_threshold=-1.5,
sell_threshold=1.5):
Strategy.__init__(self)
self.symbol = symbol
self.lookback_intervals = lookback_intervals
self.buy_threshold = buy_threshold
self.sell_threshold = sell_threshold
self.prices = pd.DataFrame()
self.is_long, self.is_short = False, False
def event_position(self, positions):
if self.symbol in positions:
position = positions[self.symbol]
self.is_long = True if position.net > 0 else False
self.is_short = True if position.net < 0 else False
def event_tick(self, market_data):
self.store_prices(market_data)
if len(self.prices) < self.lookback_intervals:
return
signal_value = self.calculate_z_score()
timestamp = market_data.get_timestamp(self.symbol)
if signal_value < self.buy_threshold:
self.on_buy_signal(timestamp)
elif signal_value > self.sell_threshold:
self.on_sell_signal(timestamp)
def store_prices(self, market_data):
timestamp = market_data.get_timestamp(self.symbol)
self.prices.loc[timestamp, "close"] = \
market_data.get_last_price(self.symbol)
self.prices.loc[timestamp, "open"] = \
market_data.get_open_price(self.symbol)
def calculate_z_score(self):
self.prices = self.prices[-self.lookback_intervals:]
returns = self.prices["close"].pct_change().dropna()
z_score = ((returns-returns.mean())/returns.std())[-1]
return z_score
def on_buy_signal(self, timestamp):
if not self.is_long:
self.send_market_order(self.symbol, 100,
True, timestamp)
def on_sell_signal(self, timestamp):
if not self.is_short:
self.send_market_order(self.symbol, 100,
False, timestamp)
最后定義一個回測類，將上述模塊串聯到一起。回測系統中只是對策略交易的已實現收益(未實現收益)進行回測，并未加入收益率、夏普比率、最大回撤等策略評價指標。
import datetime as dt
import pandas as pd
class Backtester:
def __init__(self, symbol, start_date, end_date):
self.target_symbol = symbol
self.start_dt = start_date
self.end_dt = end_date
self.strategy = None
self.unfilled_orders = []
self.positions = dict()
self.current_prices = None
self.rpnl, self.upnl = pd.DataFrame(), pd.DataFrame()
def get_timestamp(self):
return self.current_prices.get_timestamp(self.target_symbol)
def get_trade_date(self):
timestamp = self.get_timestamp()
return timestamp.strftime("%Y-%m-%d")
def update_filled_position(self, symbol, qty, is_buy,price, timestamp):
position = self.get_position(symbol)
position.event_fill(timestamp, is_buy, qty, price)
self.strategy.event_position(self.positions)
self.rpnl.loc[timestamp, "rpnl"] = position.realized_pnl
print (self.get_trade_date(), \
"成交:", "買入" if is_buy else "賣出", \
qty, symbol, "價格", price)
def get_position(self, symbol):
if symbol not in self.positions:
position = Position()
position.symbol = symbol
self.positions[symbol] = position
return self.positions[symbol]
def evthandler_order(self, order):
self.unfilled_orders.append(order)
print (self.get_trade_date(), \
"收到指令:", \
"買入" if order.is_buy else "賣出", order.qty, \
order.symbol)
def match_order_book(self, prices):
if len(self.unfilled_orders) > 0:
self.unfilled_orders = \
[order for order in self.unfilled_orders
if self.is_order_unmatched(order, prices)]
def is_order_unmatched(self, order, prices):
symbol = order.symbol
timestamp = prices.get_timestamp(symbol)
if order.is_market_order and timestamp > order.timestamp:
# Order is matched and filled.
order.is_filled = True
open_price = prices.get_open_price(symbol)
order.filled_timestamp = timestamp
order.filled_price = open_price
self.update_filled_position(symbol,
order.qty,
order.is_buy,
open_price,
timestamp)
self.strategy.event_order(order)
return False
return True
def evthandler_tick(self, prices):
self.current_prices = prices
self.strategy.event_tick(prices)
self.match_order_book(prices)
def start_backtest(self):
self.strategy = MeanRevertingStrategy(self.target_symbol)
self.strategy.event_sendorder = self.evthandler_order
mds = MarketDataSource()
mds.event_tick = self.evthandler_tick
mds.ticker = self.target_symbol
mds.start, mds.end = self.start_dt, self.end_dt
print ("Backtesting started...")
mds.start_market_simulation()
print ("Completed.")
開始回測
backtester = Backtester("600000",'20180101','20200323')
backtester.start_backtest()
Backtesting started...
2019-01-31 收到指令: 賣出 100 600000
2019-02-01 成交: 賣出 100 600000 價格 10.82
2019-02-15 收到指令: 買入 100 600000
2019-02-18 成交: 買入 100 600000 價格 10.75
......
2020-03-09 收到指令: 買入 100 600000
2020-03-10 成交: 買入 100 600000 價格 10.71
Completed.
backtester.rpnl.plot(figsize=(16,6))
plt.show()
策略已實現收益：
03 結語
本文以均值回歸模型為例，展示了基于事件驅動回測系統的Python實現過程。當然，上述回測系統仍然是一個簡化版的系統，還存在很多需要完善的地方，比如沒有加入關于策略的量化評價指標和可視化模塊，沒有考慮交易手續費等，這些都留待讀者自己去思考和進一步完善。現實的市場交易比回測系統要復雜更多，因此回測系統再怎么完美也很難完全復現真實交易的場景。量化回測是量化交易的重要組成部分，回測系統的好壞會直接影響對策略的評估。目前很多量化平臺和Python量化回測開源框架都提供了相應的回測系統，大家也沒必要都自己去重新造輪子，但是對于Python基礎比較扎實，從事個人量化交易的來說，了解回測系統的運作過程，構建自己的量化交易系統還是很有必要的。后續推文將會介紹Python量化回測開源框架的應用。
參考資料：Weiming J M, Weiming J M. Mastering Python for Finance[M]. 2015.
總結
以上是生活随笔為你收集整理的手把手教你python实现量价形态选股知乎_【手把手教你】Python实现基于事件驱动的量化回测...的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： python dict遍历文件_pyt
- 下一篇： python中的if not语句_pyt