事件驅動的雙均線策略示例

from foehnquant.base import *
from foehnquant.event import *
from foehnquant.const import *


class Strategy:

    def __init__(self, config_file, platform, symbol, timeframe, asset, fast_length, slow_length, long_stop, short_stop):
        self.realtrade = RealTrade(config_file, platform, symbol, timeframe)
        self.kline = self.realtrade.get_kline()  # 啟動時獲取一次k線
        self.kline.pop(-1)  # 去除最新的那根k線資料
        self.timestamp = None  # 設定一個變數用來判斷k線的更新
        self.cross_over, self.cross_down = False, False  # 雙均線交叉信號
        # 策略啟動時使用rest api獲取一次當前持倉資訊
        self.current_contracts = self.realtrade.current_contracts
        self.current_price = self.realtrade.current_price
        self.current_direction = self.realtrade.current_direction
        self.asset = asset  # 初始資金
        self.contract_value = self.realtrade.contract_value  # 獲取合約面值
        self.filter = False  # 過濾器，控制平倉或止損后當根bar不再下單
        self.long_stop, self.short_stop = long_stop, short_stop
        self.fast_length, self.slow_length = fast_length, slow_length
        Quant(
            config_file=config_file,
            platform=platform,
            channels=["trade", "kline.{}".format(timeframe), "order", "position", "asset"],
            symbols=[symbol],
            orderbook_update_callback=self.on_event_orderbook_update,   # orderbook資料更新回呼函式
            kline_update_callback=self.on_event_kline_update,           # k線資料更新回呼函式
            trade_update_callback=self.on_event_trade_update,           # 逐筆成交資料更新回呼函式
            order_update_callback=self.on_event_order_update,           # 訂單資料更新回呼函式
            position_update_callback=self.on_event_position_update,     # 持倉更新回呼函式
            asset_update_callback=self.on_event_asset_update            # 資產資料更新回呼函式
        )
        logger.info("策略初始化成功，開始運行...")

    def on_event_orderbook_update(self, orderbook: Orderbook):
        """訂單簿更新"""
        logger.debug("orderbook: {}".format(orderbook))

    def on_event_kline_update(self, kline: Kline):
        """k線更新"""
        logger.debug("kline: {}".format(kline))
        if kline.timestamp != self.timestamp:
            self.filter = False     # 過濾器重置為False
            self.timestamp = kline.timestamp
            # 計算信號
            self.kline.append([kline.timestamp, kline.open, kline.high, kline.low, kline.close, kline.volume])
            self.kline.pop(0)   # 增加新的k線后去除最舊的一根k線資料，優化記憶體
            ma = self.realtrade.ma(self.fast_length, self.slow_length, kline=self.kline)    # 傳入指定的k線資料
            fast_ma = ma[0]
            slow_ma = ma[1]
            self.cross_over = fast_ma[-2] > slow_ma[-2] and fast_ma[-3] <= slow_ma[-3]
            self.cross_down = fast_ma[-2] < slow_ma[-2] and fast_ma[-3] >= slow_ma[-3]
            logger.info("k線更新！上根bar短均線值：{} 上根bar長均線值：{} 上上根bar短均線值：{} 上上根bar長均線值：{} 金叉：{} 死叉：{}".format(
                fast_ma[-2], slow_ma[-2], fast_ma[-3], slow_ma[-3], self.cross_over, self.cross_down
            ))

    def on_event_trade_update(self, trade: Trade):
        """成交資料更新"""
        logger.debug("trade: {}".format(trade))
        try:
            if self.cross_over and self.filter is False:
                if self.current_contracts == 0:  # 若無持倉，買入開多
                    self.realtrade.buy(trade.price, int(self.asset / trade.price / self.contract_value), order_type="MARKET")
                    self.filter = True  # 過濾器設定為True，當根bar不再下單開倉
                elif self.current_direction == "short":
                    cover_amount = self.current_contracts
                    hold_price = self.current_price
                    result = self.realtrade.buytocover(trade.price, self.current_contracts, order_type="MARKET")  # 買入平空
                    avg_price = result['成交均價']
                    profit = (avg_price - hold_price) * cover_amount * self.contract_value
                    self.asset += profit
                    self.realtrade.buy(trade.price, int(self.asset / trade.price / self.contract_value), order_type="MARKET")
                    self.filter = True  # 過濾器設定為True，當根bar不再下單開倉
            if self.cross_down and self.filter is False:
                if self.current_contracts == 0:
                    self.realtrade.sellshort(trade.price, int(self.asset / trade.price / self.contract_value), order_type="MARKET")
                    self.filter = True  # 過濾器設定為True，當根bar不再下單開倉
                elif self.current_direction == "long":
                    cover_amount = self.current_contracts
                    hold_price = self.current_price
                    result = self.realtrade.sell(trade.price, self.current_contracts, order_type="MARKET")
                    avg_price = result['成交均價']
                    profit = (hold_price - avg_price) * cover_amount * self.contract_value
                    self.asset += profit
                    self.realtrade.sellshort(trade.price, int(self.asset / trade.price / self.contract_value), order_type="MARKET")
                    self.filter = True  # 過濾器設定為True，當根bar不再下單開倉
            # 止損
            if self.current_contracts > 0:  # 如果當前持倉數量大于0
                if self.current_direction == "long" and trade.price <= self.current_price * self.long_stop:
                    cover_amount = self.current_contracts
                    hold_price = self.current_price
                    result = self.realtrade.sell(trade.price, self.current_contracts, order_type="MARKET")
                    avg_price = result['成交均價']
                    profit = (hold_price - avg_price) * cover_amount * self.contract_value
                    self.asset += profit
                    self.filter = True  # 過濾器設定為True，當根bar不再下單開倉
                if self.current_direction == "short" and trade.price >= self.current_price * self.short_stop:
                    cover_amount = self.current_contracts
                    hold_price = self.current_price
                    result = self.realtrade.buytocover(trade.price, self.current_contracts, order_type="MARKET")  # 買入平空
                    avg_price = result['成交均價']
                    profit = (avg_price - hold_price) * cover_amount * self.contract_value
                    self.asset += profit
                    self.filter = True  # 過濾器設定為True，當根bar不再下單開倉
        except:
            logger.error()

    def on_event_order_update(self, order: Order):
        """訂單更新"""
        logger.debug("order: {}".format(order))
        if order.status == FILLED:
            push(
                "【交易訂單更新】平臺：{} 幣對：{} 方向：{} 成交數量：{} 成交均價：{} 事件時間：{}".format(
                    order.platform, order.symbol, order.action, order.filled_qty, order.avg_price, order.utime
                )
            )

    def on_event_position_update(self, position: Position):
        """持倉更新"""
        logger.debug("position: {}".format(position))
        # 持倉更新時更新持倉資訊
        if position.long_quantity > 0:
            self.current_direction = "long"
            self.current_contracts = int(position.long_quantity)    # OKEX合約以張為單位，必須是整數
            self.current_price = position.long_avg_price
            logger.info("多頭持倉更新！當前持多，數量：{} 均價：{}".format(self.current_contracts, self.current_price))
        elif position.short_quantity > 0:
            self.current_direction = "short"
            self.current_contracts = int(position.short_quantity)
            self.current_price = position.short_avg_price
            logger.info("空頭持倉更新！當前持空，數量：{} 均價：{}".format(self.current_contracts, self.current_price))
        elif position.long_quantity == 0 and position.short_quantity == 0:
            self.current_direction = "none"
            self.current_contracts = 0
            self.current_price = 0
            logger.info("持倉更新！當前無持倉！")

    def on_event_asset_update(self, asset: Asset):
        """資產更新"""
        logger.debug("asset: {}".format(asset))


if __name__ == '__main__':

    Strategy(
        config_file='config.json',
        platform=OKEXFUTURES,
        symbol="TRX-USDT-210326",
        timeframe="3m",
        asset=70,
        fast_length=20,
        slow_length=30,
        long_stop=0.95,
        short_stop=1.05
    )