Trong hai bài viết trước đây của loạt bài, chúng tôi đã thảo luận về hệ thống backtesting dựa trên sự kiện là gì và phân cấp lớp đối tượng Event.
Một trong những mục tiêu của chúng tôi với một hệ thống giao dịch dựa trên sự kiện là giảm thiểu sự trùng lặp mã giữa yếu tố backtesting và yếu tố thực thi trực tiếp. Lý tưởng nhất là sẽ tối ưu để sử dụng cùng một phương pháp tạo tín hiệu và các thành phần quản lý danh mục đầu tư cho cả thử nghiệm lịch sử và giao dịch trực tiếp. Để điều này hoạt động đối tượng Chiến lược tạo ra các tín hiệu và đối tượng danh mục đầu tư cung cấp các lệnh dựa trên chúng, phải sử dụng một giao diện giống hệt với nguồn cấp dữ liệu thị trường cho cả lịch sử và hoạt động trực tiếp.
Điều này thúc đẩy khái niệm phân cấp lớp dựa trên đối tượng DataHandler, cung cấp cho tất cả các lớp con một giao diện để cung cấp dữ liệu thị trường cho các thành phần còn lại trong hệ thống.
Các lớp con ví dụ cụ thể có thể bao gồm HistoricCSVDataHandler, QuandlDataHandler, SecuritiesMasterDataHandler, InteractiveBrokersMarketFeedDataHandler v.v. Trong hướng dẫn này, chúng tôi sẽ chỉ xem xét việc tạo một trình xử lý dữ liệu CSV lịch sử, sẽ tải dữ liệu CSV trong ngày cho cổ phiếu trong một tập hợp thanh mở-tối thấp-gần-cao-tháng lượng-mối quan tâm mở. Điều này sau đó có thể được sử dụng để
Nhiệm vụ đầu tiên là nhập các thư viện cần thiết. Cụ thể, chúng ta sẽ nhập panda và các công cụ lớp cơ sở trừu tượng.event.pynhư được mô tả trong hướng dẫn trước:
# data.py
import datetime
import os, os.path
import pandas as pd
từ abc nhập khẩu ABCMeta, abstractmethod
từ sự kiện nhập MarketEvent DataHandler là một lớp cơ bản trừu tượng (ABC), có nghĩa là không thể tạo ví dụ trực tiếp. Chỉ có các lớp con có thể được tạo ví dụ. Lý do cho điều này là ABC cung cấp một giao diện mà tất cả các lớp con DataHandler tiếp theo phải tuân thủ, do đó đảm bảo khả năng tương thích với các lớp khác giao tiếp với chúng.
Chúng tôi sử dụngMetaclassThêm vào đó, chúng ta sử dụng trình trang trí @abstractmethod để cho Python biết rằng phương thức sẽ được ghi đè trong các lớp con (điều này giống hệt với một phương thức ảo trong C++).
Hai phương pháp quan tâm là get_latest_bars và update_bars. phương pháp đầu tiên trả về N thanh cuối cùng từ dấu thời gian nhịp tim hiện tại, hữu ích cho các tính toán lăn cần thiết trong các lớp Chiến lược. phương pháp sau cung cấp một cơ chế
# data.py
class DataHandler(object):
"""
DataHandler is an abstract base class providing an interface for
all subsequent (inherited) data handlers (both live and historic).
The goal of a (derived) DataHandler object is to output a generated
set of bars (OLHCVI) for each symbol requested.
This will replicate how a live strategy would function as current
market data would be sent "down the pipe". Thus a historic and live
system will be treated identically by the rest of the backtesting suite.
"""
__metaclass__ = ABCMeta
@abstractmethod
def get_latest_bars(self, symbol, N=1):
"""
Returns the last N bars from the latest_symbol list,
or fewer if less bars are available.
"""
raise NotImplementedError("Should implement get_latest_bars()")
@abstractmethod
def update_bars(self):
"""
Pushes the latest bar to the latest symbol structure
for all symbols in the symbol list.
"""
raise NotImplementedError("Should implement update_bars()")
Với DataHandler ABC được chỉ định, bước tiếp theo là tạo một trình xử lý cho các tệp CSV lịch sử.
Máy xử lý dữ liệu yêu cầu một vài thông số, cụ thể là hàng đợi sự kiện để đẩy thông tin MarketEvent, đường dẫn tuyệt đối của tệp CSV và danh sách các ký hiệu.
# data.py
class HistoricCSVDataHandler(DataHandler):
"""
HistoricCSVDataHandler is designed to read CSV files for
each requested symbol from disk and provide an interface
to obtain the "latest" bar in a manner identical to a live
trading interface.
"""
def __init__(self, events, csv_dir, symbol_list):
"""
Initialises the historic data handler by requesting
the location of the CSV files and a list of symbols.
It will be assumed that all files are of the form
'symbol.csv', where symbol is a string in the list.
Parameters:
events - The Event Queue.
csv_dir - Absolute directory path to the CSV files.
symbol_list - A list of symbol strings.
"""
self.events = events
self.csv_dir = csv_dir
self.symbol_list = symbol_list
self.symbol_data = {}
self.latest_symbol_data = {}
self.continue_backtest = True
self._open_convert_csv_files()
Nó sẽ cố gắng mở các tệp với định dạng
Một trong những lợi ích của việc sử dụng panda như một kho dữ liệu bên trong HistoricCSVDataHandler là các chỉ mục của tất cả các biểu tượng được theo dõi có thể được hợp nhất với nhau. Điều này cho phép các điểm dữ liệu bị thiếu được lăn về phía trước, ngược hoặc can thiệp trong các khoảng trống này để có thể so sánh các dấu hiệu trên cơ sở thanh-đối-bạch. Điều này là cần thiết cho các chiến lược đảo ngược trung bình, ví dụ. Lưu ý việc sử dụng các phương pháp liên minh và tái chỉ mục khi kết hợp các chỉ mục cho tất cả các biểu tượng:
# data.py
def _open_convert_csv_files(self):
"""
Opens the CSV files from the data directory, converting
them into pandas DataFrames within a symbol dictionary.
For this handler it will be assumed that the data is
taken from DTN IQFeed. Thus its format will be respected.
"""
comb_index = None
for s in self.symbol_list:
# Load the CSV file with no header information, indexed on date
self.symbol_data[s] = pd.io.parsers.read_csv(
os.path.join(self.csv_dir, '%s.csv' % s),
header=0, index_col=0,
names=['datetime','open','low','high','close','volume','oi']
)
# Combine the index to pad forward values
if comb_index is None:
comb_index = self.symbol_data[s].index
else:
comb_index.union(self.symbol_data[s].index)
# Set the latest symbol_data to None
self.latest_symbol_data[s] = []
# Reindex the dataframes
for s in self.symbol_list:
self.symbol_data[s] = self.symbol_data[s].reindex(index=comb_index, method='pad').iterrows()
Phương thức _get_new_bar tạo ra một trình tạo để cung cấp một phiên bản định dạng của dữ liệu thanh. Điều này có nghĩa là các cuộc gọi tiếp theo đến phương thức sẽ tạo ra một thanh mới cho đến khi kết thúc dữ liệu ký hiệu:
# data.py
def _get_new_bar(self, symbol):
"""
Returns the latest bar from the data feed as a tuple of
(sybmbol, datetime, open, low, high, close, volume).
"""
for b in self.symbol_data[symbol]:
yield tuple([symbol, datetime.datetime.strptime(b[0], '%Y-%m-%d %H:%M:%S'),
b[1][0], b[1][1], b[1][2], b[1][3], b[1][4]])
Phương pháp trừu tượng đầu tiên từ DataHandler được thực hiện là get_latest_bars. Phương pháp này đơn giản cung cấp một danh sách các thanh N cuối cùng từ cấu trúc dữ liệu latest_symbol_data.
# data.py
def get_latest_bars(self, symbol, N=1):
"""
Returns the last N bars from the latest_symbol list,
or N-k if less available.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print "That symbol is not available in the historical data set."
else:
return bars_list[-N:]
Phương thức cuối cùng, update_bars là phương thức trừu tượng thứ hai từ DataHandler. Nó chỉ đơn giản tạo ra một MarketEvent được thêm vào hàng đợi khi nó thêm các thanh mới nhất vào latest_symbol_data:
# data.py
def update_bars(self):
"""
Pushes the latest bar to the latest_symbol_data structure
for all symbols in the symbol list.
"""
for s in self.symbol_list:
try:
bar = self._get_new_bar(s).next()
except StopIteration:
self.continue_backtest = False
else:
if bar is not None:
self.latest_symbol_data[s].append(bar)
self.events.put(MarketEvent())
Do đó, chúng ta có một đối tượng có nguồn gốc từ DataHandler, được sử dụng bởi các thành phần còn lại để theo dõi dữ liệu thị trường.
Trong bài viết tiếp theo chúng ta sẽ xem xét hệ thống phân cấp lớp Strategy và mô tả cách một chiến lược có thể được thiết kế để xử lý nhiều biểu tượng, do đó tạo ra nhiều SignalEvents cho đối tượng Portfolio.