【原】AI量化框架：我們?yōu)楹我匦略燧喿樱?/span>

AI量化實(shí)驗(yàn)室 2023-10-12 發(fā)布于北京

展開(kāi)全文

原創(chuàng)文章第239篇，專注“個(gè)人成長(zhǎng)與財(cái)富自由、世界運(yùn)作的邏輯與投資"。

01 為何要造輪子

星球群里，今天大家在討論一個(gè)問(wèn)題：這么多量化框架和平臺(tái)，為何還要自己造一個(gè)輪子，應(yīng)該把精力放在選股、擇時(shí)、因子、策略等等才對(duì)呀？

這個(gè)討論很有意思。其實(shí)我也不斷告訴我自己，不能陷入工程師思維，為造輪子而造輪子。仔細(xì)回想起來(lái)，從pyalgotrade，backtrader, pybroker，還有AI量化的qlib我都深度用過(guò)，寫過(guò)策略，而且呢，還細(xì)細(xì)讀過(guò)它們的代碼。另外平臺(tái)型的產(chǎn)品如聚寬，bigquant等都嘗試過(guò)策略。

第三方平臺(tái)好處是數(shù)據(jù)全，不用自己找數(shù)據(jù)，缺點(diǎn)就是慢。對(duì)于我們要不斷嘗試策略，優(yōu)化因子的過(guò)程，這個(gè)速度是非常重要的。且不說(shuō)如果你真要實(shí)盤的策略，放在第三方平臺(tái)是肯定不放心的；還有一點(diǎn)就是你若想到了平臺(tái)未實(shí)現(xiàn)的功能，那要擴(kuò)展起來(lái)就幾乎不可能了。

再說(shuō)開(kāi)源框架，有幾個(gè)成熟的框架著實(shí)不錯(cuò)，各有各的優(yōu)點(diǎn)。

但是仍然無(wú)法完全滿足我的訴求，因此，我決定取它們所有人的精華，自己整合一個(gè)。畢竟AI量化是一輩子的事業(yè)，所以“磨刀不誤砍柴功”。

希望我們的框架做到幾個(gè)點(diǎn)：

1、一定要快，可以緩存數(shù)據(jù)、因子、模型，回測(cè)結(jié)果等，并行計(jì)算，向量化加速等。

2、支持傳統(tǒng)量化?！@一點(diǎn)qlib就不滿足。

3、兼容機(jī)器學(xué)習(xí)量化，支持WFA前向滾動(dòng)式訓(xùn)練回測(cè)?！挥衠lib滿足，pyBroker有一點(diǎn)，但它是一個(gè)symbol一個(gè)model，不符合訴求。支持調(diào)參，因子擴(kuò)展等。

4、環(huán)境可以擴(kuò)展支持強(qiáng)化學(xué)習(xí)?！壳耙仓挥衠lib支持?；蛘呦駎ensortrade，finRL-meta這樣的專用框架。

5、支持“積木式”快速策略開(kāi)發(fā)。

所以，有點(diǎn)像 pybroker+qlib+bt（支持積木式）。

02 數(shù)據(jù)加載

傳統(tǒng)回測(cè)平臺(tái)，基礎(chǔ)數(shù)據(jù)與指標(biāo)是分開(kāi)的，比如backtrader和pybroker。但這在機(jī)器學(xué)習(xí)里就非常麻煩了，機(jī)器學(xué)習(xí)隨便就搞上百個(gè)因子，這要用indicator，那就太麻煩了，這里我們會(huì)參考qlib的dataset加載數(shù)據(jù)，自動(dòng)計(jì)算因子，以及自動(dòng)標(biāo)注數(shù)據(jù)—— 為里我們實(shí)現(xiàn)了自己的因子表達(dá)式計(jì)算引擎。

一堆因子自動(dòng)計(jì)算與數(shù)據(jù)標(biāo)注自動(dòng)完成，傳統(tǒng)量化框架需要一個(gè)個(gè)添加indicator，這對(duì)于機(jī)器學(xué)習(xí)而言，太麻煩，不好維護(hù)。

03 架構(gòu)設(shè)計(jì)

Portfolio有點(diǎn)類似我之前的Account的定位，但叫Portfolio更加合適，說(shuō)白了包括了cash和投資組合。做多:buy/sell，做空：short/cover。好在我們不是做通過(guò)框架，當(dāng)前市場(chǎng)我們不需要做空，先可以不支持。止損單先期都可以不支持。

ExecContext是執(zhí)行上下文，有Portfolio的實(shí)例引用，有curr_bar=傳統(tǒng)量化需要用當(dāng)天的curr_ba_df, curr_df=當(dāng)天及之前所有的bar，這個(gè)風(fēng)險(xiǎn)平價(jià)計(jì)算權(quán)重時(shí)需要用，或者滾動(dòng)機(jī)器學(xué)習(xí)時(shí)需要用。

ExecContext產(chǎn)生交易信號(hào)，所有都產(chǎn)生完成后，統(tǒng)計(jì)執(zhí)行訂單。這里寫策略就不必關(guān)心下單順序，比如“先賣后買”。

之前我的簡(jiǎn)化模型，不太心收盤價(jià)，只關(guān)心收益率，這樣做非常簡(jiǎn)單。但就沒(méi)有辦法計(jì)算某一個(gè)trade，成本是多少，持倉(cāng)多久，交易勝率之類的，因此，這里把pybroker里的邏輯學(xué)習(xí)過(guò)來(lái)，也不復(fù)雜。

# 當(dāng)日收盤合，要根據(jù)se_bar更新一次及市值，再進(jìn)行交易——次日開(kāi)盤交易（這里有滑點(diǎn)）。
def update_bar(self, date: np.datetime64, se_bar: pd.Series):

    # 當(dāng)前的cash部分
    total_equity = self.cash
    total_market_value = total_equity
    total_margin = 0.0

    for s, pos in self.long_positions.items():
        close = None
        # 這里不同市場(chǎng)，比如海外市場(chǎng)，可能不存在的，不存在變化率就是0.0， 即不變
        if s in se_bar.index:
            close = se_bar[s]

        if close:
            pos.equity = pos.shares * close
            pos.market_value = pos.equity
            pos.close = close
            pos.pnl = _calculate_pnl(close, pos.entries, "long")
        total_equity += pos.equity
        total_market_value += pos.equity

    self.equity = total_equity
    self.market_value = total_market_value
    self.margin = total_margin

    bar = PortfolioBar(
        date=date,
        cash=self.cash,
        equity=self.equity,
        market_value=self.market_value,
        margin=self.margin,
        pnl=self.market_value - self._initial_market_value,
        # fees=self.fees,
    )
    self.bars.append(bar)

交易環(huán)境上下文，不像pyBroker分成多個(gè)context，直接一個(gè)context，傳入當(dāng)前可獲得的所有數(shù)據(jù)：

像線上回測(cè)平臺(tái)那個(gè)order_target_percent這種好用的函數(shù)，是一定要實(shí)現(xiàn)的：

import numpy as np

from quant.portfolio import Portfolio
import numpy as np
import pandas as pd


class ExecContext:
    def __init__(self,
                 date: np.datetime64,
                 portfolio: Portfolio,
                 bar_df: pd.Series,
                 hist_df: pd.DataFrame
                 ):
        self.date = date
        self.portfolio = portfolio
        self.bar_df = bar_df
        self.hist_df = hist_df

    def _get_symbol_price(self, symbol):
        price = self.bar_df.loc[symbol]['close']
        return price

    def _calc_target_shares(self, symbol, target_percent) -> float:
        cash = self.portfolio.cash
        price = self._get_symbol_price(symbol)
        shares = int(cash / price)
        return shares

    def order_target_percent(self, symbol, target_percent):
        target_shares = self._calc_target_shares(symbol, target_percent)
        if symbol in self.portfolio.long_positions.keys():
            shares = self.portfolio.long_positions[symbol].shares
            if target_shares > shares:
                self.portfolio.buy(self.date, symbol, target_shares-shares
                                   , self._get_symbol_price(symbol))
            else:
                self.portfolio.sell(self.date, symbol, shares - target_shares
                                    , self._get_symbol_price(symbol))

        else:
            self.portfolio.buy(self.date, symbol, target_shares, self._get_symbol_price(symbol))