-
Notifications
You must be signed in to change notification settings - Fork 0
/
momentum v5
154 lines (129 loc) · 6.15 KB
/
momentum v5
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
import pandas as pd
import numpy as np
from scipy import stats
from heapq import nlargest
LONG_WINDOW = 100
SHORT_WINDOW = 20
NUMBER_OF_TOKENS = 10
close_prices = pd.read_csv("usdt_price_data.csv")
close_prices['timestamp'] = pd.to_datetime(close_prices['timestamp']) # Convert to datetime if it's not already
close_prices.set_index('timestamp', inplace=True)
# PARAMETERS
# how many days' data we'll look at?
lookback = 35
# ignoring last x days' data
last_days = 0
# how many days will we hold the coins?
holding_days = 7
# if BTC return is below the threshold over a given period, we hold BTC; otherwise, we buy altcoins
threshold = 0.05
btc_price = close_prices.iloc[:,0]
def slope(ts):
ts = ts.dropna()
x = np.arange(len(ts))
log_ts = np.log(ts)
slope, intercept, r_value, p_value, std_err = stats.linregress(x, log_ts)
annualized_slope = (np.power(np.exp(slope), 365) - 1) * 100
return annualized_slope * (r_value ** 2)
def information_discreteness(df):
returns = df.pct_change()
positive_returns = returns[returns > 0]
negative_returns = returns[returns < 0]
negative_positive_diff = negative_returns.count() / len(df) - positive_returns.count() / len(df)
cumul_return = df[-1] / df[0] - 1
sign = lambda x: 1 if x > 0 else -1 if x < 0 else 0
info_discreteness = sign(cumul_return) * negative_positive_diff
return info_discreteness
def calculate_beta(df, coin_column, window=30):
# Calculate daily returns
btc_returns = df.iloc[:,0].pct_change() # Bitcoin is first column
coin_returns = df[coin_column].pct_change()
# Remove any NaN values
valid_data = pd.concat([btc_returns, coin_returns], axis=1).dropna()
# Calculate covariance and variance
covariance = valid_data.iloc[:,0].cov(valid_data.iloc[:,1])
variance = valid_data.iloc[:,0].var()
# Calculate beta
beta = covariance / variance
return beta
def beta_adjust_return(return_value, beta):
"""
Adjust return to beta=1 level:
- If beta > 1, delever the return
- If beta < 1, lever the return
"""
if beta == 0: # Avoid division by zero
return return_value
return return_value / beta
def crypto_momentum(df, lookback, last_days, holding_days, threshold, commission = 0.001):
weekly_returns = []
for i in range(lookback, len(df)-lookback+1, holding_days):
if btc_price[i] / btc_price[i-lookback] - 1 > threshold:
total = 0
return_ranks = dict()
discreteness_ranks = dict()
combined_ranks = dict()
# Calculate both metrics for all valid tokens
for col in df.columns[1:]:
if np.isnan(df[col][i]) == False:
try:
# Calculate raw return over the lookback period
raw_return = df[col][i-last_days] / df[col][i-lookback] - 1
# Calculate beta for this period
beta = calculate_beta(df.iloc[i-lookback:i-last_days], col)
# Adjust return based on beta
adjusted_return = beta_adjust_return(raw_return, beta)
token_discreteness = information_discreteness(df[col][i-lookback:i-last_days])
return_ranks[col] = adjusted_return
discreteness_ranks[col] = token_discreteness
except:
pass
# Get rankings for both metrics
returns_sorted = sorted(return_ranks.items(), key=lambda x: x[1], reverse=True)
discreteness_sorted = sorted(discreteness_ranks.items(), key=lambda x: x[1], reverse=True)
# Assign ranks (1 is best)
for rank, (token, _) in enumerate(returns_sorted, 1):
combined_ranks[token] = rank
for rank, (token, _) in enumerate(discreteness_sorted, 1):
combined_ranks[token] = combined_ranks.get(token, 0) + rank
# Select tokens with lowest combined rank (best in both metrics)
five_largest = nlargest(NUMBER_OF_TOKENS, combined_ranks.keys(), key=lambda x: -combined_ranks[x])
for coin in five_largest:
try:
# For the actual portfolio return calculation, we use raw returns
weekly_return = (df[coin][i+holding_days-1] * (1-commission)) / (df[coin][i] * (1+commission)) - 1
total += weekly_return
except:
pass
avg_weekly_return = total / NUMBER_OF_TOKENS
weekly_returns.append(avg_weekly_return)
else:
avg_weekly_return = 0
weekly_returns.append(avg_weekly_return)
return [weekly_returns, five_largest]
wr = crypto_momentum(close_prices, lookback, last_days, holding_days, threshold)[0]
selected_coins = crypto_momentum(close_prices, lookback, last_days, holding_days, threshold)[1]
wr = [x for x in wr if str(x) != 'nan']
def geom_return(returns):
returns= [i + 1 for i in returns]
cumulative_returns = np.cumprod(returns)
geometric_return = cumulative_returns[-1] ** (1/len(cumulative_returns)) - 1
annualized_return = (1 + geometric_return) ** (365/holding_days) -1
return annualized_return
annualized_return = geom_return(wr)
print("Annual return is " + "{:.2%}".format(annualized_return))
def benchmark_return(df, commission=0.001):
btc_return = df.iloc[:,0][-1] * (1 - commission) / (df.iloc[:,0][0] * (1 + commission)) - 1
annual_btc_return = (1 + btc_return) ** (365 / len(df)) - 1
return annual_btc_return
annual_btc_return = benchmark_return(close_prices)
print("Benchmark return is " + "{:.2%}".format(annual_btc_return))
def calculate_max_drawdown(returns):
returns = [i+1 for i in returns]
cumulative_returns = np.cumprod(returns)
peak = np.maximum.accumulate(cumulative_returns)
drawdown = (cumulative_returns - peak) / peak
max_drawdown = np.min(drawdown)
return max_drawdown
max_drawdown = calculate_max_drawdown(wr)
print("Maximum Drawdown:", "{:.2%}".format(max_drawdown))