去掉底层回测引擎，完全自研，增加超参数优化，因子自动挖掘，机器模型交易。

昨天花一天时间，把回测引擎主体重写了。

性能提升不是一星半点。

确实是代码读多的好处，把各家所长精华都消化了，缺点尽量规避。

今天的计划：

1、轮动模板迁移过来。

2、补充orders ，trades细节。

3、调试stats细节。

使用examples/scripts下的脚本下载所需要的ETF日线数据（注意这里都是后复权的）。

轮动策略的模板同样简洁：

@dataclass
class TaskRolling(Task):  # 轮动策略模板
    def get_algos(self):
        return [
            self._parse_period(),
            SelectBySignal(rules_buy=self.rules_buy,
                           buy_at_least_count=self.at_least_buy,
                           rules_sell=self.rules_sell,
                           sell_at_least_count=self.at_least_sell
                           ),
            SelectTopK(factor_name=self.order_by, K=self.topK, drop_top_n=self.dropN,
                       b_ascending=self.b_ascending),
            self._parse_weights(),
            Rebalance()
        ]

算法每天盘后运行，先是信号选股，把rsrs标准分<0的过滤掉，然后按20天动量斜率排序，择期最大者持有。

策略配置如下：

from engine.engine import Engine
from engine.task import TaskRolling

task = TaskRolling()

task.benchmark = '510300.SH'
task.symbols = [ "513100.SH", "159934.SZ", "510880.SH", "159915.SZ",]

task.features = [ "roc(close,20)", "slope(close,20)", "slope_pair(high,low,18)", "zscore(rsrs_18,600)",]
task.feature_names = [ "roc_20", "slope_20", "rsrs_18", "rsrs_norm",]

task.rules_buy = []
task.rules_sell = ["rsrs_norm<0"]

task.order_by = 'slope_20'

e = Engine(task=task)
e.run()
e.stats()

策略表现：

策略代码在如下位置：

大家可以前往星球下载，

【就要提价了】AI量化实验室——2024量化投资的星辰大海

A股市场开年以来可是相当震撼。

上午表现还可以，群里一片哀嚎。

这个时候，大类资产配置的优势就体现出来了。

万物皆周期，投资尤是。

按自己的规划和交易系统走，不忘初心。

在目前的规划里，深度强化学习本身也是用于挖因子，不过，有quant4.0的版本里，端对端且跳过因子挖掘，或者说通过模型直接学习价量中的信息，这在图像识别和自然语言处理里，已经实现了。

神经网络能自动提取股票原始量价数据中的特征，实现端到端的因子挖掘和因子合成。注意力机制在多数场景下有效。

我们基于策略模板，尽量把开发量简化。

当前已经重构了大类资产配置模板，轮动模型，明天是择时模板，机器因子合成模板，属于轮动策略的一个分支，然后就是多策略组合。

之后是因子挖掘，jplearn，深度强化学习端对端，因子评价等。

还有就是超参数优化，包括暴力穷举，机器学习，遗传算法超参数优化等。

去掉底层回测引擎，完全自研，增加超参数优化，因子自动挖掘，机器模型交易等，之前分享过，但过于分散，整合成一体。

重写后的代码更简洁：

import pandas as pd

from .task import Task
from .strategy import StrategyAlgo, ExecContext
from .portfolio import Portfolio, PortfolioBar


class Engine:
    def __init__(self, task: Task):
        self.task = task
        self.df_datas = self.task.load_datas()
        self.dates = list(self.df_datas.index.unique())
        self.dates.sort()

        self.portfolio = Portfolio(init_cash=task.init_cash)
        self.strategy = StrategyAlgo(self.task.get_algos(), self.portfolio)
        self.symbols = list(self.df_datas['symbol'].unique())
        self.exec_context = ExecContext()
        self.exec_context.dates = self.dates
        self.exec_context.strategy = self.strategy
        self.exec_context.df_datas = self.df_datas

    def _get_bar(self, date):
        df_bar = self.df_datas.loc[date, :].copy()
        if type(df_bar) is pd.Series:
            df_bar = df_bar.to_frame().T
        df_bar.index = df_bar['symbol']
        return df_bar

    def run(self, **kwargs):  # 这里的参数用作超参数优化
        for i, date in enumerate(self.dates):
            df_bar = self._get_bar(date)

            self.exec_context.temp = {}
            self.exec_context.now = date
            self.exec_context.index = i
            self.exec_context.df_bar = df_bar

            # 先更新portfolio
            self.portfolio.on_bar(date, df_bar)
            # 收盘后交易
            self.strategy.on_bar(self.exec_context)
            self.portfolio.process_orders()

    def optimize(self):
        pass

    def stats(self):
        portfolio_df = pd.DataFrame.from_records(
            self.portfolio.bars, columns=PortfolioBar._fields, index="date"
        )
        portfolio_df['market_value'].plot()
        import matplotlib.pyplot as plt

        from matplotlib import rcParams
        rcParams['font.family'] = 'SimHei'
        portfolio_df['market_value'].plot()
        plt.show()

核心类就是portfolio，

from dataclasses import dataclass, field
from typing import NamedTuple

import numpy as np
import pandas as pd
from loguru import logger


class ScheOrder(NamedTuple):
    symbol: str
    amount: float


@dataclass
class Position:
    symbol: str
    shares: float
    close: float  # 最近的收盘价
    equity: float
    bars: int = 0


class PortfolioBar(NamedTuple):
    date: np.datetime64
    cash: float
    equity: float
    market_value: float
    fees: float  # 手续费


class Portfolio:
    def __init__(self, init_cash, fee_rate=0.000):
        self.positions: dict[str, Position] = {}
        self.cash = init_cash
        self.fees = 0.0
        self.bars: list[PortfolioBar] = list()
        self.sche_orders = []
        self.curr_bar = None
        self.fee_rate = fee_rate
        self.total_market_value = init_cash

    def on_bar(self, date: np.datetime64, df_bar: pd.DataFrame):
        total_equity = 0.0
        self.curr_bar = df_bar

        for symbol in self.positions.keys():
            se = df_bar.loc[symbol]
            pos = self.positions[symbol]
            pos.close = se['close']
            pos.equity = pos.shares * pos.close  # 这里更新equity。
            self.positions[symbol] = pos

            total_equity += pos.equity
        self.total_market_value = total_equity + self.cash

        self.bars.append(PortfolioBar(
            date=date,
            cash=self.cash,
            equity=total_equity,
            market_value=self.total_market_value,
            fees=self.fees
        ))

    # strategy下单在这里，统一执行，先卖再买
    def new_order(self, symbol, amount):
        order = ScheOrder(symbol=symbol, amount=amount)
        if order.amount < 0:
            self.sche_orders.insert(0, order)
        else:
            self.sche_orders.append(order)

    def process_orders(self):
        for o in self.sche_orders:
            price = self.curr_bar.loc[o.symbol]['close']
            shares = int(o.amount / price)
            if o.amount > 0:
                self._buy(o.symbol, shares)
            else:
                self._sell(o.symbol, shares)
        self.sche_orders.clear()

    def _buy(self, symbol, shares):
        if shares < 1:
            return
        #assert shares >= 1
        if symbol not in self.curr_bar.index:
            logger.error('当天{}没有数据'.format(symbol))
            return

        price = self.curr_bar.loc[symbol]['close']
        amount = price * shares
        fee = amount * self.fee_rate
        total_amount = amount + fee

        if self.cash < total_amount:
            logger.error('现金不够，无法下单：{}'.format(symbol))
            return

        if symbol in self.positions.keys():
            pos = self.positions[symbol]
            pos.shares += shares
        else:
            pos = Position(symbol=symbol, shares=shares, close=price, equity=shares*price)
        self.positions[symbol] = pos
        self.cash -= total_amount

    def _sell(self, symbol, shares):
        if symbol not in self.curr_bar.index:
            logger.error('当天{}没有数据'.format(symbol))
            return

        price = self.curr_bar.loc[symbol]['close']
        amount = price * shares

        if symbol in self.positions.keys():
            pos = self.positions[symbol]
            if pos.shares < shares:
                logger.error('当前持仓股数：{}小于{}，交易无法进行'.format(pos.shares, shares))
                return
            pos.shares -= shares
            if pos.shares == 0:
                del self.positions[symbol]

            fee = amount * self.fee_rate
            self.cash += (amount-fee)
        else:
            logger.error('当前未持仓，无法卖出:{}'.format(symbol))