engine重构：新增order_amount函数

昨天完成了海龟策略的出入场信息。

但没有写仓位管理的逻辑。出入场的信号就是通道突破。进场的仓位管理，继续加仓以及动态止损。满足出场或止损信号说清仓。

仓位管理中一个交易系统中重要的风控的一环。无法是传统规则量化，还是AI机器学习预测，都是一个概率、胜率的问题，而风险控制则是确保我们的策略持有体验，能够坚持下去的重要一环，毕竟，交易最重要的事情，是安全，然后是盈利！

我们在交易记录里添加了price，记录交易时候的价格：

在engine的如下文件：

price=self._get_symbol_price(symbol),

# 交易记录def _record_orders(self, date, old_pos: dict, new_pos: dict):    for symbol, _ in old_pos.items():        if symbol not in new_pos.keys():  # 清仓了            order_type = 'sell'            trade_amount = old_pos[symbol]        else:            new_mv = new_pos[symbol]            old_mv = old_pos[symbol]            trade_amount = abs(new_mv - old_mv)            if trade_amount == 0:                continue            if new_mv > old_mv:                order_type = 'buy'            else:                order_type = 'sell'        fees = _calc_fees(trade_amount)        self.fees += fees        self._order_id += 1        self.orders.append(Order(date=date,                                 symbol=symbol,                                 price=self._get_symbol_price(symbol),                                 market_value=trade_amount,                                 type=order_type,                                 fees=fees,                                 id=self._order_id                                 ))

今天对engine核心回测引擎做了较大升级：

重点函数order_amount（symbol, amount)。就是对某证券买入一定的市值，这个做基金的同学会比较熟悉，我们定投，网格都是按市值买，多如买入5万块的创业板ETF或者卖出5000块的纳指ETF。

——我们很少关心ETF本身的价格，以及买了多少份。

这一点与股票不同，作为通行的回测系统，我们不太关心一手（100股）这些细节，比如BTC那一枚可能好几万，回测系统怎么办对不对，就是按市值来，你可以买0.001份，这样就通用了。

from collections import defaultdict, deque
from typing import NamedTuple, Literal, List

import numpy as np
import pandas as pd
from loguru import logger
from engine.config import g
from dataclasses import dataclass, field


def _calc_fees(trade_amount):
    return trade_amount * 0.0001


class Order(NamedTuple):
    id: int
    symbol: str
    type: Literal["buy", "sell"]
    date: np.datetime64
    price: float  # 成交价格
    market_value: float
    fees: float


class PortfolioBar(NamedTuple):
    date: np.datetime64  # 日期
    cash: float  # 当天收盘且交易之后的现金
    equity: float  # 当天收盘且交易之后的权益（市值-现金）
    market_value: float  # 总市值
    pnl: float  # 当前总损益
    fees: float  # 当前默计手续费


class Portfolio:
    def __init__(self, init_cash: float = 100000.0):
        self.curr_bar_df: pd.DataFrame = None  # 当前最新的一个bar_df
        self.init_cash = init_cash
        self.curr_cash = init_cash
        self.curr_date = None
        self.curr_holding = defaultdict(float)  # 当前持仓{symbol:权益市值}
        self.bars = []
        self.orders = []
        self.symbol_orders = defaultdict(list)  # 历史交易订单，按symbol记录
        self.fees = 0.0
        self._order_id = 0

        # 当日收盘合，要根据se_bar更新一次及市值，再进行交易——次日开盘交易（这里有滑点）。

    def update_bar(self, date: np.datetime64, se_bar: pd.Series, df_bar: pd.DataFrame):
        self.curr_bar_df = df_bar
        # 所有持仓的，按收益率更新mv
        total_equity = 0.0
        # 当前已经持仓中的symbol，使用收盘后的收益率更新
        for s, equity in self.curr_holding.items():
            rate = 0.0
            # 这里不同市场，比如海外市场，可能不存在的，不存在变化率就是0.0， 即不变
            if s in se_bar.index:
                rate = se_bar[s]
            new_equity = equity * (1 + rate)
            self.curr_holding[s] = new_equity
            total_equity += new_equity

        bar = PortfolioBar(date=date,
                           cash=self.curr_cash,
                           equity=total_equity,
                           market_value=self.curr_cash + total_equity,
                           pnl=self.curr_cash + total_equity - self.init_cash,
                           fees=0.0
                           )
        self.bars.append(bar)
        self.curr_date = date

    def _get_symbol_price(self, symbol):
        if self.curr_bar_df is None:
            logger.error('当前bar_df为空，取不到价格')
            return None

        if symbol not in self.curr_bar_df.index:
            logger.error('{}不在bar_df的索引中，取不到价格')
            return None

        return self.curr_bar_df.loc[symbol]['close']

    def _place_order(self, symbol, amount):
        self.curr_holding[symbol] += amount
        self.curr_cash -= amount
        order_type = 'buy'
        if amount < 0:
            order_type = 'sell'

        if self.curr_holding[symbol] == 0:
            self.curr_holding.pop(symbol)

        self._order_id += 1
        o = Order(date=self.curr_date,
                  symbol=symbol,
                  price=self._get_symbol_price(symbol),
                  market_value=abs(amount),
                  type=order_type,
                  fees=_calc_fees(abs(amount)),
                  id=self._order_id
                  )
        # print(o)
        self.orders.append(o)
        self.symbol_orders[symbol].append(o)

    # 按amount买入symbol，比如买入SPX:5万块，定投5000块。这个符合买基金的直觉; amount为正是买入，负为卖出
    def order_amount(self, symbol, amount):
        if symbol not in self.curr_bar_df.index:
            logger.error('日期{},{}不存在'.format(self.curr_date, symbol))
            return

            # 买入，需要看现金够不够
        if amount > 0:
            if self.curr_cash < amount:
                logger.error('日期{},{}现金不够'.format(self.curr_date, symbol))
                return
            self._place_order(symbol, amount)

        if amount < 0:  # 看持仓够不够
            if symbol not in self.curr_holding.keys():
                logger.error('{}未持仓，无法卖出')
                return

            if self.curr_holding[symbol] < amount:
                logger.warning('{}当前仅持仓：{}，清仓')
                self.close_symbol(symbol)
                return

            self._place_order(symbol, amount)

    def close_symbol(self, symbol):
        if symbol not in self.curr_holding.keys():
            #logger.error('{}未持仓，无法清仓')
            return
        holding = self.curr_holding[symbol]
        self._place_order(symbol, -holding)

    # 持仓市值，不包括cash
    def _calc_total_holding_equity(self):
        total_mv = 0.0
        for s, mv in self.curr_holding.items():
            total_mv += mv
        return total_mv

    # == 一些接口
    def get_total_mv(self):
        return self._calc_total_holding_equity() + self.curr_cash

扩展函数移动了 context.py里：

大家重点关心order_target_weights，order_taget_amount。

import math
from typing import Optional

import numpy as np

from engine.portfolio import Portfolio, Order
from engine.broker import Broker, ScheOrder
import numpy as np
import pandas as pd
from loguru import logger


class ExecContext:
    def __init__(self,
                 index: int,
                 date: np.datetime64,
                 portfolio: Portfolio,
                 bar_df: pd.DataFrame,
                 hist_df: Optional[pd.DataFrame] = None,
                 bench_hist_df: Optional[pd.DataFrame] = None,
                 ):
        self.index = index
        self.date = date
        self.portfolio = portfolio
        self.bar_df = bar_df
        self.hist_df = hist_df
        self.bench_hist_df = bench_hist_df
        self.algo_context = {}

    def order_target_amount(self, symbol, amount):
        if amount < 0:
            logger.error('目标仓位必须大于0')
            return
        old_amount = self.portfolio.curr_holding[symbol]
        delta = amount - old_amount
        if delta < 10:  # 市值基本不变，不调
            return
        self.portfolio.order_amount(symbol, delta)

    def order_target_weight(self, symbol, weight):
        if weight < 0 or weight > 1:
            logger.error('目标权重必须在[0-1]之间')
            return
        total_mv = self.portfolio.get_total_mv()
        target_mv = total_mv * weight
        self.order_target_amount(symbol, target_mv)

    # 这里负责调仓顺序，先卖后买
    def order_target_weights(self, weights: dict):
        total_mv = self.portfolio.get_total_mv()
        curr_weights = {s: (mv / total_mv) for s, mv in self.portfolio.curr_holding.items()}

        #卖出的先调仓
        to_sell = []
        to_close = []
        for holding, _ in self.portfolio.curr_holding.items():
            if holding not in weights.keys():
                to_close.append(holding)

        self.close_symbols(to_close)

        to_buy = []
        for s, new_w in weights.items():
            if s not in curr_weights.keys() or new_w < curr_weights[s]:
                to_sell.append(s)
            else:
                to_buy.append(s)

        for s in to_sell:
            self.order_target_weight(s, weights[s])

        for s in to_buy:
            self.order_target_weight(s, weights[s])

    def close_all(self):
        symbols = self.bar_df.index
        self.close_symbols(symbols)

    def close_symbols(self, symbols):
        for s in symbols:
            self.portfolio.close_symbol(s)

    def get_long_symbols(self):
        symbols_holding = []
        for s, mv in self.portfolio.curr_holding.items():
            if mv > 0:
                symbols_holding.append(s)
        return symbols_holding

    # 获取上一次交易价格（如果有），没有就返回None，表明未持他
    def get_last_long_price(self, symbol):
        if symbol in self.portfolio.symbol_orders.keys():
            orders = self.portfolio.symbol_orders[symbol].copy()
            if len(orders) > 0:
                orders.reverse()
                for o in orders:
                    if o.type == 'buy':
                        return o.price
        return None