保守投资公式

1import numpy as np
2from AlgorithmImports import *
3from typing import List, Dict
4from numpy import isnan
5class TheConservativeFormula(QCAlgorithm):
6def Initialize(self):
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000)
9self.period:int = 36 * 21
10self.leverage:int = 10
11self.quantile:int = 5
12self.long:List[Symbol] = []
13self.data:Dict[Symbol, SymbolData] = {}
14
15market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
16
17self.fundamental_count:int = 500
18self.fundamental_sorting_key = lambda x: x.DollarVolume
19self.selection_flag:bool = False
20self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
21self.UniverseSettings.Resolution = Resolution.Daily
22self.AddUniverse(self.FundamentalSelectionFunction)
23self.Schedule.On(self.DateRules.MonthStart(market), self.TimeRules.AfterMarketOpen(market), self.Selection)
24self.settings.daily_precise_end_time = False
25def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
26for security in changes.AddedSecurities:
27    security.SetFeeModel(CustomFeeModel())
28    security.SetLeverage(self.leverage)
29    
30def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> List[Symbol]:
31# Update the rolling window every day.
32for stock in fundamental:
33    symbol:Symbol = stock.Symbol
34    # Store monthly price.
35    if symbol in self.data:
36        self.data[symbol].update(stock.AdjustedPrice)
37if not self.selection_flag:
38    return Universe.Unchanged
39selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Market == 'usa' and \
40                          not isnan(x.ValuationRatios.TotalYield) and (x.ValuationRatios.TotalYield > 0)]
41
42# Warmup price rolling windows.
43for stock in selected:
44    symbol:Symbol = stock.Symbol
45    if symbol in self.data:
46        continue
47    
48    self.data[symbol] = SymbolData(self.period)
49    history = self.History(symbol, self.period, Resolution.Daily)
50    if history.empty:
51        self.Log(f"Not enough data for {symbol} yet.")
52        continue
53    closes = history.loc[symbol].close
54    for time, close in closes.items():
55        self.data[symbol].update(close)
56    
57yield_performance_volatility:Dict[Symbol, (Tuple[float])] = { x.Symbol : (x.ValuationRatios.TotalYield , self.data[x.Symbol].performance(), self.data[x.Symbol].volatility()) for x in selected if self.data[x.Symbol].is_ready()}
58
59# Volatility sorting.
60if len(yield_performance_volatility) < 2:
61    return Universe.Unchanged
62sorted_by_ret_vol:List[Tuple[Symbol, float]] = sorted(yield_performance_volatility.items(), key = lambda x: x[1][2], reverse = True)
63half:int = int(len(sorted_by_ret_vol) / 2)
64low_by_ret_vol:List[Tuple[Symbol, float]] = [x for x in sorted_by_ret_vol[-half:]]
65
66# Scoring
67rank:Dict[Symbol, int] = {}
68for symbol, _ in low_by_ret_vol:
69    rank[symbol] = 0
70
71sorted_by_mom:List[Tuple[Symbol, float]] = sorted(low_by_ret_vol, key = lambda x: x[1][1], reverse = True)
72score:int = len(sorted_by_mom)
73for symbol, _ in sorted_by_mom:
74    rank[symbol] += score
75    score -= 1
76    
77sorted_by_yield:List[Tuple[Symbol, float]] = sorted(low_by_ret_vol, key = lambda x: x[1][0], reverse = True)
78score:int = len(sorted_by_yield)
79for symbol, _ in sorted_by_yield:
80    rank[symbol] += score
81    score -= 1
82if len(rank) >= self.quantile:
83    sorted_by_rank:List[Tuple[Symbol, float]] = sorted(rank.items(), key = lambda x: x[1], reverse = True)
84    quintile:int = int(len(sorted_by_rank) / self.quantile)
85
86    self.long:List[Symbol] = [x[0] for x in sorted_by_rank[:quintile]]
87
88return self.long
89def OnData(self, data: Slice) -> None:
90if not self.selection_flag: 
91    return
92self.selection_flag = False
93
94# Trade execution
95portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, 1 / len(self.long)) for symbol in self.long if symbol in data and data[symbol]]
96self.SetHoldings(portfolio, True)
97self.long.clear()
98def Selection(self) -> None:
99if self.Time.month % 3 == 0:
100    self.selection_flag = True
101class SymbolData():
102def __init__(self, period:int):
103self._price:RollingWindow = RollingWindow[float](period)
104
105def update(self, price:float) -> None:
106self._price.Add(price)
107
108def is_ready(self) -> bool:
109return self._price.IsReady
110
111def performance(self, values_to_skip = 0) -> float:
112closes:List[float] = [x for x in self._price][:12*21][values_to_skip:]
113return (closes[0] / closes[-1] - 1)                
114
115def volatility(self) -> float:
116closes:np.ndarray = np.array([x for x in self._price])
117returns:np.ndarray = (closes[:-1] / closes[1:]) - 1
118return np.std(returns)
119
120# Custom fee model.
121class CustomFeeModel(FeeModel):
122def GetOrderFee(self, parameters):
123fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
124return OrderFee(CashAmount(fee, "USD"))