-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmomentum v10
152 lines (128 loc) · 6.21 KB
/
momentum v10
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
import pandas as pd
import numpy as np
from heapq import nlargest, nsmallest
import matplotlib.pyplot as plt
NUMBER_OF_TOKENS = 5
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 = 14
# if BTC return is below the threshold over a given period, we hold BTC; otherwise, we buy altcoins
THRESHOLD = 0.00
COMMISSION = 0.001
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 trading_strategy(df, lookback=LOOKBACK, last_days=LAST_DAYS, holding_days=HOLDING_DAYS, threshold=THRESHOLD , num_tokens=NUMBER_OF_TOKENS, commission=COMMISSION):
"""
A practical trading strategy that:
1. Ranks coins by 35-day returns
2. Ranks by Information Discreteness
3. Selects top 10 coins
4. Goes long when BTC trend is favorable
"""
selected_coins_history = [] # Track selected coins for each period
portfolio_returns = []
for i in range(lookback, len(df), holding_days):
try:
# Check if market conditions are favorable
if btc_price[i] / btc_price[i-lookback] - 1 > threshold:
# Store metrics for each coin
coin_metrics = {}
# Calculate metrics for all valid tokens
for col in df.columns[1:]:
if np.isnan(df[col][i]) == False:
try:
# Calculate return
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])
coin_metrics[col] = {
'return': token_return,
'discreteness': token_discreteness
}
except:
continue
if len(coin_metrics) < num_tokens:
continue
# Sort coins by return
return_sorted = sorted(coin_metrics.items(),
key=lambda x: x[1]['return'],
reverse=True)
# Take top 30% by returns and sort them by ID
top_return_count = int(len(return_sorted) * 0.3)
top_returns = return_sorted[:top_return_count]
# Sort by information discreteness (lowest)
selected_coins = nlargest(num_tokens,
[(coin, metrics['discreteness']) for coin, metrics in top_returns],
key=lambda x: -x[1]) # Negative for lowest ID
selected_coins = [coin for coin, _ in selected_coins]
selected_coins_history.append(selected_coins)
# Calculate portfolio return
total_return = 0
valid_positions = 0
for coin in selected_coins:
try:
# Long-only strategy
coin_return = (df[coin][i+holding_days-1] * (1-commission)) / (df[coin][i] * (1+commission)) - 1
total_return += coin_return
valid_positions += 1
except:
try:
coin_return = (df[coin].iloc[-1] * (1-commission)) / (df[coin][i] * (1+commission)) - 1
total_return += coin_return
valid_positions += 1
except:
continue
if valid_positions > 0:
portfolio_return = total_return / valid_positions
portfolio_returns.append(portfolio_return)
else:
portfolio_returns.append(0)
else:
portfolio_returns.append(0)
selected_coins_history.append([])
except Exception as e:
print(f"Error at timestamp {i}: {str(e)}")
portfolio_returns.append(0)
selected_coins_history.append([])
return portfolio_returns, selected_coins_history
# Run the strategy
returns, selected_coins = trading_strategy(close_prices)
# Calculate performance metrics
returns = [x for x in returns if str(x) != 'nan']
cumulative_returns = np.cumprod([1 + r for r in returns])
total_return = cumulative_returns[-1] - 1
annualized_return = (1 + total_return) ** (365 / (len(returns) * 7)) - 1
max_drawdown = np.min((cumulative_returns - np.maximum.accumulate(cumulative_returns)) / np.maximum.accumulate(cumulative_returns))
sharpe_ratio = np.mean(returns) / np.std(returns) if np.std(returns) != 0 else 0
# Print results
print(f"\nStrategy Performance:")
print(f"Total Return: {total_return:.2%}")
print(f"Annualized Return: {annualized_return:.2%}")
print(f"Maximum Drawdown: {max_drawdown:.2%}")
print(f"Sharpe Ratio: {sharpe_ratio:.2f}")
# Plot cumulative returns
plt.figure(figsize=(12, 6))
plt.plot(cumulative_returns)
plt.title('Strategy Cumulative Returns')
plt.xlabel('Weeks')
plt.ylabel('Cumulative Return')
plt.grid(True)
plt.show()