量化交易之回测篇 - 落单品种单策略的交易记录（template.py）

import re
from abc import ABC
from copy import copy
from typing import Any, Callable

import pandas

from public_module.constant import Interval, Direction, Offset
from public_module.object import BarData, TickData, OrderData, TradeData
from public_module.utility import virtual

from public_module.base import StopOrder
from datetime import datetime

from public_module.tqz_extern.tools.file_path_operator.file_path_operator import TQZFilePathOperator
class CtaTemplate(ABC):
    """"""

    author = ""
    parameters = []
    variables = []

    back_tester_type = True

    def __init__(
        self,
        cta_engine: Any,
        strategy_name: str,
        vt_symbol: str,
        setting: dict,
    ):
        """"""
        self.cta_engine = cta_engine
        self.strategy_name = strategy_name
        self.vt_symbol = vt_symbol

        self.inited = False
        self.trading = False
        self.datetime = None
        self.trade_df = pandas.DataFrame()
        self.pos = 0

        # Copy a new variables list here to avoid duplicate insert when multiple
        # strategy instances are created with the same strategy class.
        self.variables = copy(self.variables)
        self.variables.insert(0, "inited")
        self.variables.insert(1, "trading")
        self.variables.insert(2, "pos")

        self.update_setting(setting)



    """ 中间略过本blog未涉及内容 """

    @virtual
    def on_init(self):
        """
        Callback when strategy is inited.
        """
        pass

    @virtual
    def on_start(self):
        """
        Callback when strategy is started.
        """
        pass

    @virtual
    def on_stop(self):
        """
        Callback when strategy is stopped.
        """

        bt_name = f'{self.strategy_name.split(".")[0]}{self.strategy_name.split(".")[1]}_{self.strategy_name.split(".")[2]}.xlsx'
        excel_writer = pandas.ExcelWriter(path=TQZFilePathOperator.grandfather_path(source_path=__file__) + f'/back_tester_result/stock/{bt_name}')
        self.trade_df.to_excel(excel_writer, sheet_name='trade_df', index=False, freeze_panes=(1, 0))
        excel_writer.save()

    @virtual
    def on_tick(self, tick: TickData):
        """
        Callback of new tick data update.
        """
        pass

    @virtual
    def on_bar(self, bar: BarData):
        """
        Callback of new bar data update.
        """
        self.datetime = bar.datetime



    def buy(self, price: float, volume: float, stop: bool = False, lock: bool = False):
        """
        Send buy order to open a long position.
        """
        if self.back_tester_type:
            self.pos += volume

        self.__update_trade_df(bar_dt=self.datetime, price=price, bs='卖', oc='平', unit_pos=volume)

        return self.send_order(Direction.LONG, Offset.OPEN, price, volume, stop, lock)



    def sell(self, price: float, volume: float, stop: bool = False, lock: bool = False):
        """
        Send sell order to close a long position.
        """
        if self.back_tester_type:
            self.pos -= volume

        self.__update_trade_df(bar_dt=self.datetime, price=price, bs='卖', oc='平', unit_pos=volume)

        return self.send_order(Direction.SHORT, Offset.CLOSE, price, volume, stop, lock)



    def short(self, price: float, volume: float, stop: bool = False, lock: bool = False):
        """
        Send short order to open as short position.
        """
        if self.back_tester_type:
            self.pos -= volume

        self.__update_trade_df(bar_dt=self.datetime, price=price, bs='卖', oc='开', unit_pos=volume)

        return self.send_order(Direction.SHORT, Offset.OPEN, price, volume, stop, lock)



    def cover(self, price: float, volume: float, stop: bool = False, lock: bool = False):
        """
        Send cover order to close a short position.
        """

        if self.back_tester_type:
            self.pos += volume

        self.__update_trade_df(bar_dt=self.datetime, price=price, bs='买', oc='平', unit_pos=volume)

        return self.send_order(Direction.LONG, Offset.CLOSE, price, volume, stop, lock)



    def __update_trade_df(self, bar_dt: datetime, price: float, bs: str, oc: str, unit_pos: float):

        current_row = len(self.trade_df)
        self.trade_df.loc[current_row, 'date'] = bar_dt
        self.trade_df.loc[current_row, '买/卖'] = bs
        self.trade_df.loc[current_row, '开/平'] = oc
        self.trade_df.loc[current_row, 'price'] = price
        self.trade_df.loc[current_row, 'unit_pos'] = unit_pos