-
Notifications
You must be signed in to change notification settings - Fork 0
/
momentum v7
172 lines (144 loc) · 7.46 KB
/
momentum v7
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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
import pandas as pd
import numpy as np
from heapq import nlargest, nsmallest
import matplotlib.pyplot as plt
NUMBER_OF_TOKENS = 10
NUM_BUCKETS = 4
close_prices = pd.read_csv("usdt_price_data.csv")
close_prices['timestamp'] = pd.to_datetime(close_prices['timestamp'])
close_prices.set_index('timestamp', inplace=True)
# some coins should be deleted because they are stablecoins or fiat currencies
deleted_coins = ['PAX', 'BUSD', 'USDP', 'FDUSD', 'USDC', 'UST', 'TUSD', 'SUSD', 'EUR', 'EURI', 'GBP', 'AUD', 'PAXG']
close_prices = close_prices.drop(columns=deleted_coins, errors='ignore')
# 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 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 double_sort_momentum(df, lookback, last_days, holding_days, threshold, num_buckets=NUM_BUCKETS, coins_per_bucket=NUMBER_OF_TOKENS, commission=0.001):
weekly_returns = {i: [] for i in range(num_buckets)} # Store returns for each bucket
for i in range(lookback, len(df), holding_days):
try:
if btc_price[i] / btc_price[i-lookback] - 1 > threshold:
# Dictionaries to store metrics
returns_dict = {}
discreteness_dict = {}
# Calculate both metrics for all valid tokens
for col in df.columns[1:]:
if np.isnan(df[col][i]) == False:
try:
# Calculate return over the lookback period
token_return = df[col][i-last_days] / df[col][i-lookback] - 1
# Calculate information discreteness
token_discreteness = information_discreteness(df[col][i-lookback:i-last_days])
returns_dict[col] = token_return
discreteness_dict[col] = token_discreteness
except:
pass
if len(returns_dict) < num_buckets * coins_per_bucket:
continue
# Sort tokens by returns
sorted_by_returns = sorted(returns_dict.items(), key=lambda x: x[1], reverse=True)
# Create buckets based on returns
bucket_size = len(sorted_by_returns) // num_buckets
buckets = {}
# Distribute tokens into buckets
for bucket in range(num_buckets):
start_idx = bucket * bucket_size
end_idx = start_idx + bucket_size if bucket < num_buckets - 1 else len(sorted_by_returns)
bucket_tokens = sorted_by_returns[start_idx:end_idx]
# For each bucket, sort tokens by information discreteness
bucket_tokens_dict = {token: discreteness_dict[token] for token, _ in bucket_tokens}
top_tokens = nsmallest(coins_per_bucket, bucket_tokens_dict.items(), key=lambda x: x[1])
buckets[bucket] = [token for token, _ in top_tokens]
# Calculate returns for each bucket
for bucket, tokens in buckets.items():
total_return = 0
valid_tokens = 0
for token in tokens:
try:
weekly_return = (df[token][i+holding_days-1] * (1-commission)) / (df[token][i] * (1+commission)) - 1
total_return += weekly_return
valid_tokens += 1
except:
# If we're near the end of the dataset, use the last available price
try:
weekly_return = (df[token].iloc[-1] * (1-commission)) / (df[token][i] * (1+commission)) - 1
total_return += weekly_return
valid_tokens += 1
except:
continue
if valid_tokens > 0:
avg_bucket_return = total_return / valid_tokens
weekly_returns[bucket].append(avg_bucket_return)
else:
weekly_returns[bucket].append(0)
else:
# If BTC threshold not met, append 0 returns for all buckets
for bucket in range(num_buckets):
weekly_returns[bucket].append(0)
except:
break
return weekly_returns, buckets
def analyze_bucket_performance(weekly_returns_dict):
bucket_performance = {}
for bucket, returns in weekly_returns_dict.items():
returns = [x for x in returns if str(x) != 'nan']
if returns:
# Calculate cumulative return
cumulative_return = np.prod([1 + r for r in returns]) - 1
# Calculate annualized return
ann_return = (1 + cumulative_return) ** (365 / (len(returns) * 7)) - 1
# Calculate max drawdown
cumulative_returns = np.cumprod([1 + r for r in returns])
peak = np.maximum.accumulate(cumulative_returns)
drawdown = (cumulative_returns - peak) / peak
max_drawdown = np.min(drawdown)
bucket_performance[bucket] = {
'annualized_return': ann_return,
'max_drawdown': max_drawdown,
'sharpe_ratio': np.mean(returns) / np.std(returns) if np.std(returns) != 0 else 0
}
return bucket_performance
def calculate_btc_drawdown(df):
btc_prices = df['BTC'].values
peak = np.maximum.accumulate(btc_prices)
drawdown = (btc_prices - peak) / peak
max_drawdown = np.min(drawdown)
return max_drawdown
# Run the strategy
weekly_returns_by_bucket, final_buckets = double_sort_momentum(close_prices, lookback, last_days, holding_days, threshold)
performance = analyze_bucket_performance(weekly_returns_by_bucket)
btc_drawdown = calculate_btc_drawdown(close_prices)
# Print results
print("\nBitcoin Maximum Drawdown:", "{:.2%}".format(btc_drawdown))
for bucket, metrics in performance.items():
print(f"\nBucket {bucket} Performance:")
print(f"Annualized Return: {metrics['annualized_return']:.2%}")
print(f"Maximum Drawdown: {metrics['max_drawdown']:.2%}")
print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
# Plot cumulative returns for each bucket
plt.figure(figsize=(12, 6))
for bucket, returns in weekly_returns_by_bucket.items():
returns = [x for x in returns if str(x) != 'nan']
cumulative_returns = np.cumprod([1 + r for r in returns])
plt.plot(cumulative_returns, label=f'Bucket {bucket}')
plt.legend()
plt.title('Cumulative Returns by Bucket')
plt.xlabel('Weeks')
plt.ylabel('Cumulative Return')
plt.show()