-
Notifications
You must be signed in to change notification settings - Fork 0
/
momentum v4
120 lines (102 loc) · 5.03 KB
/
momentum v4
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
import pandas as pd
import numpy as np
from scipy import stats
from heapq import nlargest
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 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[2:]:
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
token_discreteness = information_discreteness(df[col][i-lookback:i-last_days])
return_ranks[col] = raw_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))