Economic Momentum in Currencies

1from AlgorithmImports import *
2from io import StringIO
3import data_tools
4from dateutil.relativedelta import relativedelta
5import numpy as np
6import pandas as pd
7from collections import deque
8from pandas.core.frame import DataFrame
9# endregion
10class EconomicMomentuminCurrencies(QCAlgorithm):
11def Initialize(self):
12self.SetStartDate(2005, 1, 1)
13self.SetCash(100000)
14self.leverage:int = 5
15self.period:int = 60
16self.volatility_period:int = 36
17self.rolling_period:int = 2
18self.max_missing_days:int = 32
19self.min_lookback_period:int = 24
20self.data:Dict[Symbol, data_tools.SymbolData] = {}
21self.custom_data_df:Dict[str, DataFrame] = {}
22self.measures:Dict[Symbol, Tuple[int, float]] = {}
23self.custom_data_index:List[str] = ['IND_PRO_countries', 'goods_export_countries', 'goods_import_countries']
24self.custom_data_perc:List[str] = ['CPI_countries', 'LTIR_countries', 'STIR_countries', 'UNEMPLOYMENT_countries']
25# create dataframes from custom data
26for variable in self.custom_data_index + self.custom_data_perc:
27    load:str = self.Download(f'data.quantpedia.com/backtesting_data/economic/{variable}.csv')
28    df:DataFrame = pd.read_csv(StringIO(load), delimiter=';')
29    df['TIME'] = pd.to_datetime(df['TIME']).dt.date
30    df.set_index('TIME', inplace=True)
31    self.custom_data_df[variable] = df
32
33# calculate yield sread and trade balance of countries
34self.custom_data_df['yield_spread'] = self.custom_data_df['LTIR_countries'] - self.custom_data_df['STIR_countries']
35self.custom_data_df['trade_balance'] = (self.custom_data_df['goods_export_countries'] - self.custom_data_df['goods_import_countries']) \
36     / (self.custom_data_df['goods_export_countries'] + self.custom_data_df['goods_import_countries'])
37self.custom_data_df = {k:v for k, v in self.custom_data_df.items() if k not in ['goods_export_countries', 'goods_import_countries']}
38self.symbols:Dict[str, str] = {"AUDUSD": 'AUS', "GBPUSD": 'GBR', "CADUSD": 'CAN', "EURUSD": 'EU', "JPYUSD": 'JPN', "NOKUSD": 'NOR', "SEKUSD": 'SWE', "NZDUSD": 'NZL', "CHFUSD": 'CHE'}
39
40# data subscription
41for symbol in self.symbols:
42    data:Security = self.AddForex(symbol, Resolution.Daily, Market.Oanda)
43    data.SetFeeModel(data_tools.CustomFeeModel())
44    data.SetLeverage(self.leverage)
45    self.data[data.Symbol] = data_tools.SymbolData(self.rolling_period)
46self.selection_flag:bool = False            
47self.Settings.MinimumOrderMarginPortfolioPercentage = 0.
48self.recent_month:int = -1
49def OnData(self, data: Slice) -> None:
50# monthly rebalance
51if self.recent_month == self.Time.month:
52    return
53self.recent_month = self.Time.month
54
55# store monthly prices
56for symbol, symbol_data in self.data.items():
57    if symbol in data and data[symbol]:
58        symbol_data.update_price(data[symbol].Price)
59look_back:DateTime.date = self.Time.date() - relativedelta(months=self.period)
60current_date:DateTime.date = self.Time.date()
61currency_returns:Dict[Symbol, float] = {self.symbols[sym.Value] : data.get_return() for sym, data in self.data.items() if data.is_ready()}
62if len(currency_returns) == 0:
63    return
64returns_df:DataFrame = pd.DataFrame([currency_returns])
65for i in range(self.min_lookback_period, self.period):
66    measure_df:DataFrame = pd.DataFrame()
67    weights_df:DataFrame = pd.DataFrame()
68    for data_title, df in self.custom_data_df.items():
69        total_performance:float = 0
70        df:DataFrame = df.loc[df.index < current_date]
71        # check if current data are in dataframe
72        if (current_date - df.iloc[-1].name).days >= self.max_missing_days:
73            if data_title in self.measures:
74                self.measures.pop(data_title)
75            continue
76        
77        # shifted values for measures
78        measure:float = np.log(df.iloc[-2, df.columns.isin(list(returns_df.columns))]) - np.log(df.iloc[-i - 1, df.columns.isin(list(returns_df.columns))]) \
79                        if data_title in self.custom_data_index else df.iloc[-2, df.columns.isin(list(returns_df.columns))] - df.iloc[-i - 1, df.columns.isin(list(returns_df.columns))]
80        
81        # rank based on measures
82        df_ranks:DataFrame = pd.DataFrame([measure.rank()]).dropna(axis=1)
83        measure_df = returns_df
84        for country in df_ranks:
85            weight:float = (df_ranks[country] - df_ranks.mean(axis=1)) / ((df_ranks.max(axis=1) - df_ranks.min(axis=1)) / 2)
86            if country in returns_df:
87                country_performance = returns_df[country] * weight
88                total_performance += country_performance
89            weights_df[country] = weight
90        if data_title not in self.measures:
91            self.measures[data_title] = {}
92        if i not in self.measures[data_title]:
93            self.measures[data_title][i] = deque(maxlen=self.volatility_period)
94        self.measures[data_title][i].append((total_performance, weights_df))
95if self.Time.month % 3 != 0: 
96    return
97traded_portfolio_portion:Dict[str, float] = {}
98if any([any([len(data) == data.maxlen for lb, data in value.items()]) for key, value in self.measures.items()]):
99    aggregated_inverse_volatility:float = sum([sum([1 / np.std(list(map(lambda x:x[0][0], v))) for i,v in y.items()]) for x, y in self.measures.items()])
100    for variable, lookback in self.measures.items():
101        for lb, perf_weights in lookback.items():
102            weight:float = (1 / np.std(list(map(lambda x:x[0][0], list(perf_weights))))) / aggregated_inverse_volatility
103            for country in perf_weights[-1][1]:
104                portion:float = (self.Portfolio.TotalPortfolioValue / len(self.measures) / len(lookback) / len(perf_weights[-1][1])) * weight 
105                if np.isnan(portion):
106                    continue
107                if country not in traded_portfolio_portion:
108                    traded_portfolio_portion[country] = 0
109                traded_portfolio_portion[country] += portion
110# trade execution
111invested:List[Symbol] = [x.Key for x in self.Portfolio if x.Value.Invested]
112for symbol in invested:
113    if self.symbols[symbol.Value] not in traded_portfolio_portion:
114        self.Liquidate(symbol)
115for symbol in list(self.data.keys()):
116    if symbol in data and data[symbol]:
117        if self.symbols[symbol.Value] in traded_portfolio_portion:
118            quantity:float = (traded_portfolio_portion[self.symbols[symbol.Value]] // data[symbol].Price) - self.Portfolio[symbol].Quantity
119            self.MarketOrder(symbol, quantity)