quantdigger/quantdigger/engine/context.py at quantweb · midgithub/quantdigger

776 lines (678 loc) · 26.6 KB
# -*- coding: utf-8 -*-
# @file data.py
# @brief 数据上下文，交易上下文。
# @author wondereamer
# @version 0.2
# @date 2015-12-09
import copy
import datetime
import Queue
from quantdigger.engine.blotter import SimpleBlotter
from quantdigger.engine.exchange import Exchange
from quantdigger.engine.series import SeriesBase, NumberSeries, DateTimeSeries
from quantdigger.event import Event, EventsPool, SignalEvent, OnceEvent
from quantdigger.technicals.base import TechnicalBase
from quantdigger.util import elogger as logger
from quantdigger.datastruct import (
    TradeSide,
    Direction,
    PriceType,
    PositionKey,
    Contract,
class DataContext(object):
    def __init__(self, wrapper):
        data = wrapper.data
        self.open = NumberSeries(data.open.values, 'open')
        self.close = NumberSeries(data.close.values, 'close')
        self.high = NumberSeries(data.high.values, 'high')
        self.low = NumberSeries(data.low.values, 'low')
        self.volume = NumberSeries(data.volume.values, 'volume')
        self.datetime = DateTimeSeries(data.index, 'datetime')
        self.i = -1   # 第i个组合
        self.j = -1   # 第j个策略
        self.bar = Bar(None, None, None, None, None, None)
        self.last_row = []
        self.last_date = datetime.datetime(2100, 1, 1)
        self.indicators = [[{}]]
        self._curbar = -1
        self._wrapper = wrapper
        self._series = [[{}]]
        self._variables = [[{}]]
        self._all_variables = [[{}]]
        self._size = len(data.close)
    @property
    def raw_data(self):
        return self._wrapper.data
    @property
    def curbar(self):
        return self._curbar + 1
    @property
    def pcontract(self):
        return self._wrapper.pcontract
    @property
    def contract(self):
        return self._wrapper.pcontract.contract
    def __getattr__(self, name):
        return self.get_item(name)
    def update_system_vars(self):
        # self.data = np.append(data, tracker.container_day)
        self._curbar = self.last_curbar
        self.open.update_curbar(self._curbar)
        self.close.update_curbar(self._curbar)
        self.high.update_curbar(self._curbar)
        self.low.update_curbar(self._curbar)
        self.volume.update_curbar(self._curbar)
        self.datetime.update_curbar(self._curbar)
        self.bar = Bar(self.datetime[0], self.open[0], self.close[0],
                       self.high[0], self.low[0], self.volume[0])
        self.last_row = []
        return
    def rolling_forward(self):
        """ 滚动读取下一步的数据。 """
        new_row, self.last_curbar = self._wrapper.rolling_forward()
        if not new_row:
            self.last_curbar -= 1
            return False, None
        self.last_row = [1]  # mark
        self.last_date = self._wrapper.data.index[self.last_curbar]
        if self.datetime[0] >= self.last_date and self.curbar != 0:
            logger.error('合约[%s] 数据时间逆序或冗余' % self.pcontract)
            assert(False)
        return True, new_row
    def update_user_vars(self):
        """ 更新用户定义的变量。 """
        try:
            siter = self._series[self.i][self.j].iteritems()
        except IndexError:
            # The strategy doesn't have user defined series.
            pass
        else:
            for key, s in siter:
                s.update_curbar(self._curbar)
                s.duplicate_last_element()
        try:
            indic_iter = self.indicators[self.i][self.j].iteritems()
        except IndexError:
            # The strategy doesn't use indicators.
            pass
        else:
            for key, indic in indic_iter:
                if indic.is_multiple:
                    for key, value in indic.series.iteritems():
                        value.update_curbar(self._curbar)
                else:
                    for s in indic.series:
                        s.update_curbar(self._curbar)
    def __len__(self):
        return len(self._wrapper)
    def get_item(self, name):
        """ 获取用户在策略on_init函数中初始化的变量 """
        return self._all_variables[self.i][self.j][name]
    def add_item(self, name, value):
        """ 添加用户初始化的变量。 """
        # @TODO ...
        if self.i < len(self._all_variables):
            if self.j < len(self._all_variables[self.i]):
                self._all_variables[self.i][self.j][name] = value
            else:
                self._all_variables[self.i].append({name: value})
        else:
            self._all_variables.append([{name: value}])
        if isinstance(value, SeriesBase):
            self.add_series(name, value)
        elif isinstance(value, TechnicalBase):
            self.add_indicator(name, value)
        else:
            self.add_variable(name, value)
    def add_series(self, attr, s):
        """ 添加on_init中初始化的序列变量
        Args:
            attr (str): 属性名
            s (Series): 序列变量
        """
        s.reset_data([], self._size)
        if self.i < len(self._series):
            if self.j < len(self._series[self.i]):
                self._series[self.i][self.j][attr] = s
            else:
                self._series[self.i].append({attr: s})
        else:
            self._series.append([{attr: s}])
        return
    def add_indicator(self, attr, indic):
        if self.i < len(self.indicators):
            if self.j < len(self.indicators[self.i]):
                self.indicators[self.i][self.j][attr] = indic
            else:
                self.indicators[self.i].append({attr: indic})
        else:
            self.indicators.append([{attr: indic}])
    def add_variable(self, attr, var):
        if self.i < len(self._variables):
            if self.j < len(self._variables[self.i]):
                self._variables[self.i][self.j][attr] = var
            else:
                self._variables[self.i].append({attr: var})
        else:
            self._variables.append([{attr: var}])
class DataContextWrapper(object):
    def __init__(self):
        self.data = None
    def __setattr__(self, name, value):
        if name != 'data':
            data = self.data
            if name in data._all_variables[data.i][data.j]:
                data.add_item(name, value)
                return
        super(DataContextWrapper, self).__setattr__(name, value)
    def __getattr__(self, name):
        return getattr(self.data, name)
class StrategyContext(object):
    """ 策略组合
    :ivar name: 策略名
    :ivar blotter: 订单管理
    :ivar exchange: 价格撮合器
    :ivar marks: 绘图标志集合
    def __init__(self, name, settings={}):
        self.events_pool = EventsPool()
        # @TODO merge blotter and exchange
        self.blotter = SimpleBlotter(name, self.events_pool, settings)
        self.exchange = Exchange(name, self.events_pool, strict=True)
        self.name = name
        # 0: line_marks, 1: text_marks
        self.marks = [{}, {}]
        self._entry_orders = []
        self._exit_orders = []
        self._datetime = None
        self._cancel_now = False  # 是当根bar还是下一根bar撤单成功。
    def update_environment(self, dt, ticks, bars):
        """ 更新模拟交易所和订单管理器的数据，时间,持仓 """
        self.blotter.update_datetime(dt)
        self.exchange.update_datetime(dt)
        self.blotter.update_data(ticks, bars)
        self._datetime = dt
        return
    def process_trading_events(self, at_baropen):
        """ 提交订单，撮合，更新持仓 """
        append = at_baropen
        entry_flag = True
        exit_flag = True
        if self._exit_orders:
            self.events_pool.put(SignalEvent(self._exit_orders))
            self._process_trading_events(at_baropen, at_baropen)
            self._exit_orders = []
            exit_flag = False
            append = False
        if self._entry_orders:
            self.events_pool.put(SignalEvent(self._entry_orders))
            self._process_trading_events(at_baropen, append)
            self._entry_orders = []
            entry_flag = False
            append = False
        # 没有交易信号，确保至少运行一次
        if exit_flag and entry_flag:
            self.events_pool.put(OnceEvent())
            self._process_trading_events(at_baropen, append)
    def plot_line(self, name, ith_window, x, y, styles, lw=1, ms=10, twinx=False):
        """ 绘制曲线
        Args:
            name (str): 标志名称
            ith_window (int): 在第几个窗口显示，从1开始。
            x (datetime): 时间坐标
            y (float): y坐标
            styles (str): 控制颜色，线的风格，点的风格
            lw (int): 线宽
            ms (int): 点的大小
        """
        mark = self.marks[0].setdefault(name, [])
        mark.append((ith_window, twinx, x, y, styles, lw, ms))
    def plot_text(self, name, ith_window, x, y, text, color='black', size=15, rotation=0):
        """ 绘制文本
        Args:
            name (str): 标志名称
            ith_window (int): 在第几个窗口显示，从1开始。
            x (float): x坐标
            y (float): y坐标
            text (str): 文本内容
            color (str): 颜色
            size (int): 字体大小
            rotation (float): 旋转角度
        """
        mark = self.marks[1].setdefault(name, [])
        mark.append((ith_window, x, y, text, color, size, rotation))
    def _process_trading_events(self, at_baropen, append):
        """"""
        while True:
            # 事件处理。
            try:
                event = self.events_pool.get()
            except Queue.Empty:
                assert(False)
            except IndexError:
                break
            else:
                # if event.type == 'MARKET':
                    # strategy.calculate_signals(event)
                    # port.update_timeindex(event)
                if event.type == Event.SIGNAL:
                    assert(not at_baropen)
                    self.blotter.update_signal(event)
                elif event.type == Event.ORDER:
                    assert(not at_baropen)
                    self.exchange.insert_order(event)
                elif event.type == Event.FILL:
                    # 模拟交易接口收到报单成交
                    self.blotter.api.on_transaction(event)
            # 价格撮合。note: bar价格撮合要求撮合置于运算后面。
            # @TODO tick 回测不一样
            if event.type == Event.ONCE or event.type == Event.ORDER:
                self.exchange.make_market(self.blotter._bars, at_baropen)
        self.blotter.update_status(self._datetime, at_baropen, append)
        return
    def buy(self, price, quantity, price_type, contract):
        self._entry_orders.append(Order(
            None,
            contract,
            price_type,
            TradeSide.KAI,
            Direction.LONG,
            float(price),
            quantity
    def sell(self, price, quantity, price_type, contract):
        self._exit_orders.append(Order(
            None,
            contract,
            price_type,
            TradeSide.PING,
            Direction.LONG,
            float(price),
            quantity
    def short(self, price, quantity, price_type, contract):
        self._entry_orders.append(Order(
            None,
            contract,
            price_type,
            TradeSide.KAI,
            Direction.SHORT,
            float(price),
            quantity
    def cover(self, price, quantity, price_type, contract):
        self._exit_orders.append(Order(
            None,
            contract,
            price_type,
            TradeSide.PING,
            Direction.SHORT,
            float(price),
            quantity
    def cancel(self, orders):
        """ 撤单
        Args:
            orders (list/Order): 撤单，参数为list表示撤多个单。
        """
        orders = orders if isinstance(orders, list) else [orders]
        if not self._cancel_now:
            # bar close点处理撤单
            for order in orders:
                norder = copy.deepcopy(order)
                norder.side = TradeSide.CANCEL
                # @TODO or self._exit_orders ?
                # 结合实盘考虑, 实盘可能自动撤单。
                self._entry_orders.append(norder)
            return
        ## 立即处理撤单
        #temp = copy.deepcopy(self._orders)
        #self._orders = []
        #for order in orders:
            #order.side = TradeSide.CANCEL
            #self._orders.append(order)
        #self.process_trading_events(False)
        #self._orders = temp
    @property
    def open_orders(self):
        """ 未成交的订单 """
        return self.blotter.open_orders
    def all_positions(self):
        return self.blotter.positions.values()
    def position(self, contract, direction):
        try:
            poskey = PositionKey(contract, direction)
            return self.blotter.positions[poskey]
        except KeyError:
            return None
    def pos(self, contract, direction):
        try:
            poskey = PositionKey(contract, direction)
            return self.blotter.positions[poskey].closable
        except KeyError:
            return 0
    def cash(self):
        return self.blotter.holding['cash']
    def equity(self):
        return self.blotter.holding['equity']
    def profit(self, contract):
        pass
    def day_profit(self, contract):
        """ 当前持仓的浮动盈亏 """
        pass
class Context(object):
    """ 上下文"""
    def __init__(self, data, max_window):
        self.ctx_dt_series = DateTimeSeries(
            [datetime.datetime(2100, 1, 1)]*max_window,
            'universal_time')
        self.ctx_datetime = datetime.datetime(2100, 1, 1)
        self.on_bar = False
        self.step = 0
        self._data_contexts = {}       # str(PContract): DataContext
        self._contract2contexts = {}   # str(Contract): DataContext
        self._code2contexts = {}       # code: DataContext
        for key, value in data.iteritems():
            self._data_contexts[key] = value
            self._contract2contexts[key.split('-')[0]] = value
            self._code2contexts[key.split('.')[0]] = value
        self._cur_data_context = None
        self._strategy_contexts = []
        self._cur_strategy_context = None
        self._ticks = {}  # Contract: float
        self._bars = {}  # Contract: Bar
        self._data_context_wrapper = DataContextWrapper()
    def add_strategy_context(self, ctxs):
        self._strategy_contexts.append(ctxs)
    def switch_to_contract(self, pcon):
        self._cur_data_context = self._data_contexts[pcon]
    def time_aligned(self):
        return (self._cur_data_context.datetime[0] <= self.ctx_datetime and
                self._cur_data_context.last_date <= self.ctx_datetime)
        ## 第一根是必须运行
        # return  (self._cur_data_context.datetime[0] <= self.ctx_dt_series and
                # self._cur_data_context.ctx_dt_series <= self.ctx_dt_series) or \
                # self._cur_data_context.curbar == 0
    def switch_to_strategy(self, i, j, trading=False):
        self._trading = trading
        self._cur_strategy_context = self._strategy_contexts[i][j]
        if self._trading:
            for data_context in self._data_contexts.values():
                data_context.i, data_context.j = i, j
        else:
            self._cur_data_context.i, self._cur_data_context.j = i, j
    def switch_to_data(self, i, j):
        assert(False)
        self._cur_data_context.i, self._cur_data_context.j = i, j
    def process_trading_events(self, at_baropen):
        self._cur_strategy_context.update_environment(
            self.ctx_datetime, self._ticks, self._bars)
        self._cur_strategy_context.process_trading_events(at_baropen)
    def rolling_forward(self):
        """ 更新最新tick价格，最新bar价格, 环境时间。 """
        if self._cur_data_context.last_row:
            self.ctx_dt_series.curbar = self.step
            self.ctx_datetime = min(self._cur_data_context.last_date,
                                    self.ctx_datetime)
            try:
                self.ctx_dt_series.data[self.step] = min(
                    self._cur_data_context.last_date, self.ctx_datetime)
            except IndexError:
                self.ctx_dt_series.data.append(
                    min(self._cur_data_context.last_date, self.ctx_datetime))
            return True
        hasnext, data = self._cur_data_context.rolling_forward()
        if not hasnext:
            return False
        self.ctx_dt_series.curbar = self.step
        try:
            self.ctx_dt_series.data[self.step] = min(
                self._cur_data_context.last_date, self.ctx_datetime)
        except IndexError:
            self.ctx_dt_series.data.append(min(
                self._cur_data_context.last_date, self.ctx_datetime))
        self.ctx_datetime = min(
            self._cur_data_context.last_date, self.ctx_datetime)
        return True
    def update_user_vars(self):
        """ 更新用户在策略中定义的变量, 如指标等。 """
        self._cur_data_context.update_user_vars()
    def update_system_vars(self):
        """ 更新用户在策略中定义的变量, 如指标等。 """
        self._cur_data_context.update_system_vars()
        self._ticks[self._cur_data_context.contract] = \
            self._cur_data_context.close[0]
        self._bars[self._cur_data_context.contract] = \
            self._cur_data_context.bar
        oldbar = self._bars.setdefault(
            self._cur_data_context.contract, self._cur_data_context.bar)
        if self._cur_data_context.bar.datetime > oldbar.datetime:
            # 处理不同周期时间滞后
            self._bars[self._cur_data_context.contract] = \
                self._cur_data_context.bar
    @property
    def strategy(self):
        """ 当前策略名 """
        return self._cur_strategy_context.name
    @property
    def pcontract(self):
        """ 当前周期合约 """
        return self._cur_data_context.pcontract
    @property
    def symbol(self):
        """ 当前合约 """
        return str(self._cur_data_context.pcontract.contract)
    @property
    def curbar(self):
        """ 当前是第几根k线, 从1开始 """
        if self.on_bar:
            return self.step + 1
        else:
            return self._cur_data_context.curbar
    @property
    def open(self):
        """ k线开盘价序列 """
        return self._cur_data_context.open
    @property
    def close(self):
        """ k线收盘价序列 """
        return self._cur_data_context.close
    @property
    def high(self):
        """ k线最高价序列 """
        return self._cur_data_context.high
    @property
    def low(self):
        """ k线最低价序列 """
        return self._cur_data_context.low
    @property
    def volume(self):
        """ k线成交量序列 """
        return self._cur_data_context.volume
    @property
    def datetime(self):
        """ k线时间序列 """
        if self.on_bar:
            return self.ctx_dt_series
            # return self._cur_data_context.datetime
        else:
            return self._cur_data_context.datetime
    @property
    def open_orders(self):
        """ 未成交的订单 """
        return list(self._cur_strategy_context.open_orders)
    def __getitem__(self, strpcon):
        """ 获取跨品种合约 """
        # @TODO 字典，数字做key表序号, 合并key
        # @TODO 根据exception自动判断优先级
        strpcon = strpcon.upper()
        try:
            # xxxxx.xx
            self._data_context_wrapper.data = self._contract2contexts[strpcon]
        except KeyError:
            try:
                # xxxxx
                self._data_context_wrapper.data = self._code2contexts[strpcon]
            except KeyError:
                # xxxxx.xx-xx.xx
                self._data_context_wrapper.data = self._data_contexts[strpcon]
        return self._data_context_wrapper
    def __getattr__(self, name):
        return self._cur_data_context.get_item(name)
    def __setattr__(self, name, value):
        if name in [
                '_data_contexts', '_cur_data_context', '_cur_strategy_context',
                '_strategy_contexts', 'ctx_dt_series', '_ticks', '_bars',
                '_trading', 'on_bar', 'step', 'ctx_datetime',
                '_contract2contexts', '_code2contexts', '_data_context_wrapper'
            super(Context, self).__setattr__(name, value)
        else:
            self._cur_data_context.add_item(name, value)
    def buy(self, price, quantity, symbol=None):
        """ 开多仓
        Args:
            price (float): 价格, 0表市价。
            quantity (int): 数量。
            symbol (str): 合约
        """
        if not self._trading:
            raise Exception('只有on_bar函数内能下单！')
        if symbol:
            contract = Contract(symbol) if isinstance(symbol, str) else symbol
        else:
            contract = self._cur_data_context.contract
        price_type = PriceType.MKT if price == 0 else PriceType.LMT
        self._cur_strategy_context.buy(price,
                                       quantity, price_type,
    def sell(self, price, quantity, symbol=None):
        """ 平多仓。
        Args:
           price (float): 价格, 0表市价。
           quantity (int): 数量。
           symbol (str): 合约
        """
        if not self._trading:
            raise Exception('只有on_bar函数内能下单！')
        if symbol:
            contract = Contract(symbol) if isinstance(symbol, str) else symbol
        else:
            contract = self._cur_data_context.contract
        price_type = PriceType.MKT if price == 0 else PriceType.LMT
        self._cur_strategy_context.sell(price,
                                        quantity, price_type,
    def short(self, price, quantity, symbol=None):
        """ 开空仓
        Args:
            price (float): 价格, 0表市价。
            quantity (int): 数量。
            symbol (str): 合约
        """
        if not self._trading:
            raise Exception('只有on_bar函数内能下单！')
        if symbol:
            contract = Contract(symbol) if isinstance(symbol, str) else symbol
        else:
            contract = self._cur_data_context.contract
        price_type = PriceType.MKT if price == 0 else PriceType.LMT
        self._cur_strategy_context.short(price,
                                         quantity, price_type,
    def cover(self, price, quantity, symbol=None):
        """ 平空仓。
        Args:
           price (float): 价格, 0表市价。
           quantity (int): 数量。
           symbol (str): 合约
        """
        if not self._trading:
            raise Exception('只有on_bar函数内能下单！')
        if symbol:
            contract = Contract(symbol) if isinstance(symbol, str) else symbol
        else:
            contract = self._cur_data_context.contract
        price_type = PriceType.MKT if price == 0 else PriceType.LMT
        self._cur_strategy_context.cover(price,
                                         quantity, price_type,
    def position(self, direction='long', symbol=None):
        """ 当前仓位。
        Args:
            direction (str/int): 持仓方向。多头 - 'long' / 1 ；空头 - 'short'  / 2
            , 默认为多头。
            symbol (str): 字符串合约，默认为None表示主合约。
        Returns:
            Position. 该合约的持仓
        """
        if not self._trading:
            raise Exception('只有on_bar函数内能查询当前持仓！')
        direction = Direction.arg_to_type(direction)
        contract = Contract(symbol) if symbol else \
            self._cur_data_context.contract
        # @TODO assert direction
        return self._cur_strategy_context.position(contract, direction)
    def all_positions(self):
        return self._cur_strategy_context.all_positions()
    def pos(self, direction='long', symbol=None):
        """  合约的当前可平仓位。
        Args:
            direction (str/int): 持仓方向。多头 - 'long' / 1 ；空头 - 'short'  / 2
            , 默认为多头。
            symbol (str): 字符串合约，默认为None表示主合约。
        Returns:
            int. 该合约的持仓数目。
        """
        if not self._trading:
            raise Exception('只有on_bar函数内能查询当前持仓！')
        direction = Direction.arg_to_type(direction)
        # @TODO symbol xxxxx
        contract = Contract(symbol) if symbol else \
            self._cur_data_context.contract
        # @TODO assert direction
        return self._cur_strategy_context.pos(contract, direction)
    def cancel(self, orders):
        """ 撤单 """
        self._cur_strategy_context.cancel(orders)
    def cash(self):
        """ 现金。 """
        if not self._trading:
            raise Exception('只有on_bar函数内能查询可用资金！')
        return self._cur_strategy_context.cash()
    def equity(self):
        """ 当前权益 """
        if not self._trading:
            raise Exception('只有on_bar函数内能查询当前权益！')
        return self._cur_strategy_context.equity()
    def profit(self, contract=None):
        """ 当前持仓的历史盈亏 """
        #if not self._trading:
            #logger.warn('只有on_bar函数内能查询总盈亏！')
            #return
        pass
    def plot_line(self, name, ith_window, x, y, styles, lw=1, ms=10, twinx=False):
        self._cur_strategy_context.plot_line(name, ith_window-1, x-1, float(y),
                styles, lw, ms, twinx)
    def plot_text(self, name, ith_window, x, y, text, color='black', size=15, rotation=0):
        self._cur_strategy_context.plot_text(name, ith_window-1, x-1, float(y),
                                                text, color, size, rotation)
    def day_profit(self, contract=None):
        """ 当前持仓的浮动盈亏 """
        #if not self._trading:
            #logger.warn('只有on_bar函数内能查询浮动盈亏！')
            #return
        pass
    def test_cash(self):
        """  当根bar时间终点撮合后的可用资金，用于测试。 """
        self.process_trading_events(at_baropen=False)
        return self.cash()
    def test_equity(self):
        """  当根bar时间终点撮合后的权益，用于测试。 """
        self.process_trading_events(at_baropen=False)
        return self.equity()
Provide feedback

Saved searches

Use saved searches to filter your results more quickly

FilesExpand file tree

context.py

Latest commit

History

context.py

File metadata and controls
