|
原创文章第275篇，专注“个人成长与财富自由、世界运作的逻辑与投资"。 昨天完成了海龟策略的出入场信息。 海龟策略：一个完整的交易系统的实现（源码）
但没有写仓位管理的逻辑。出入场的信号就是通道突破。进场的仓位管理，继续加仓以及动态止损。满足出场或止损信号说清仓。 仓位管理中一个交易系统中重要的风控的一环。无法是传统规则量化，还是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
知识星球与开源项目：万物之中，希望至美
年化41.4%的指数多因子轮动与年化26.5%大类资产动量轮动，准备实盘跟踪。（代码下载）
|
