Investment Effect in China

1from AlgorithmImports import *
2from numpy import isnan
3
4class InvestmentEffectInChina(QCAlgorithm):
5
6def Initialize(self):
7self.SetStartDate(2000, 1, 1)
8self.SetCash(100000)
9
10self.data:Dict[Symbol, SymbolData] = {}
11self.weight:Dict[Symbol, float] = {}
12
13self.percentile_size:int = 5
14self.portfolio_size:float = 0.1
15
16self.TA_period:int = 2
17self.TA_growth_period:int = 2
18self.leverage:int = 5
19
20self.market:Symbol = self.AddEquity('SPY', Resolution.Daily).Symbol
21
22self.rebalance_flag:bool = False
23self.selection_flag:bool = False
24self.rebalance_months:List[int] = [5, 11]
25self.selection_months:List[int] = [2, 5, 8, 11]
26self.UniverseSettings.Resolution = Resolution.Daily
27self.AddUniverse(self.FundamentalSelectionFunction)
28self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
29self.Schedule.On(self.DateRules.MonthStart(self.market), self.TimeRules.BeforeMarketClose(self.market), self.Selection)
30
31def OnSecuritiesChanged(self, changes: SecurityChanges) -> None:
32for security in changes.AddedSecurities:
33    security.SetFeeModel(CustomFeeModel())
34    security.SetLeverage(self.leverage)
35
36def FundamentalSelectionFunction(self, fundamental: List[Fundamental]) -> None:
37if not self.selection_flag:
38    return Universe.Unchanged
39self.selection_flag = False
40
41# filter stocks, which symbols isn't SPY equity symbol and has fundamental data
42selected:List[Fundamental] = [x for x in fundamental if x.HasFundamentalData and x.Symbol != self.market and x.CompanyReference.BusinessCountryID == 'CHN' and \
43    not isnan(x.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths) and x.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths != 0 and x.MarketCap != 0
44    ]
45
46AAG:Dict[Fundamental, float] = {}
47
48for stock in selected:
49    symbol:Symbol = stock.Symbol
50    total_assets:float = stock.FinancialStatements.BalanceSheet.TotalAssets.ThreeMonths
51   
52    if symbol not in self.data or symbol not in self.previous_fine:
53        self.data[symbol] = SymbolData(self.TA_period, self.TA_growth_period)
54    
55    symbol_obj:SymbolData = self.data[symbol]  
56    symbol_obj.update_TA(total_assets)
57    
58    # update Total Assets growth only in May and November
59    if self.Time.month in self.rebalance_months:
60        if symbol_obj.TA_growths_ready() and symbol_obj.TA_ready():
61            # calculate stock's AAG
62            curr_TA_growth:float = symbol_obj.calculate_TA_growth()
63            prev_TA_growths:List[float] = [x for x in symbol_obj._total_assets_growths]
64            
65            AAG_value:float = curr_TA_growth - np.mean(prev_TA_growths)
66            
67            AAG[stock] = AAG_value
68            
69            # update stock's total assets growth 
70            symbol_obj.update_TA_growths(curr_TA_growth)
71            
72        elif symbol_obj.TA_ready():
73            # update stock's total assets growth
74            curr_TA_growth:float = symbol_obj.calculate_TA_growth()
75            symbol_obj.update_TA_growths(curr_TA_growth)
76    
77self.previous_fine = list(map(lambda stock: stock.Symbol, selected))
78  
79# make sure there are enough data for selection
80if len(AAG) < self.percentile_size:
81    return Universe.Unchanged
82    
83percentile:int = int(len(AAG) / self.percentile_size)
84sorted_by_AAG:List[Fundamental] = [x[0] for x in sorted(AAG.items(), key=lambda item: item[1])]
85
86# long the highest
87long = sorted_by_AAG[-percentile:]
88
89# short the lowest
90short = sorted_by_AAG[:percentile]
91
92# calculate total capitalization for long and short part
93for i, portfolio in enumerate([long, short]):
94    mc_sum:float = sum([x.MarketCap for x in portfolio])
95    for stock in portfolio:
96        self.weight[symbol] = ((-1) ** i) * stock.MarketCap / mc_sum * self.portfolio_size
97
98return list(self.weight.keys()) 
99
100def OnData(self, data: Slice) -> None:
101# rebalancing only in May and November
102if not self.rebalance_flag:
103    return
104self.rebalance_flag = False
105
106# trade execution
107portfolio:List[PortfolioTarget] = [PortfolioTarget(symbol, w) for symbol, w in self.weight.items() if symbol in data and data[symbol]]
108self.SetHoldings(portfolio, True)
109
110self.weight.clear()
111
112def Selection(self) -> None:
113# perform selection on February, May, August and November
114if self.Time.month in self.selection_months:
115    self.selection_flag = True
116    
117# perform rebalance in April, May, October and November 
118if self.Time.month in self.rebalance_months:
119    self.rebalance_flag = True
120
121class SymbolData():
122def __init__(self, TA_period: int, TA_growth_period: int) -> None:
123self._total_assets:RollingWindow = RollingWindow[float](TA_period)
124self._total_assets_growths:RollingWindow = RollingWindow[float](TA_growth_period)
125
126def update_TA(self, total_assets: float) -> None:
127self._total_assets.Add(total_assets)
128
129def update_TA_growths(self, total_assets_growths) -> None:
130self._total_assets_growths.Add(total_assets_growths)
131
132def calculate_TA_growth(self) -> float:
133total_assets:List[float] = list(self._total_assets)
134return (total_assets[0] - total_assets[-1]) / total_assets[-1] 
135
136def TA_ready(self) -> bool:
137return self._total_assets.IsReady
138
139def TA_growths_ready(self) -> bool:
140return self._total_assets_growths.IsReady
141
142# Custom fee model
143class CustomFeeModel(FeeModel):
144def GetOrderFee(self, parameters):
145fee = parameters.Security.Price * parameters.Order.AbsoluteQuantity * 0.00005
146return OrderFee(CashAmount(fee, "USD"))