Розробка моделі кластеризації криптовалют за поведінкою

Розробка моделі кластеризації крипто-активів за поведінкою

Кластеризація криптовалют за поведінковими паттернами дозволяє автоматично групувати активи зі схожими характеристиками. Це корисно для: побудови диверсифікованих портфелів, пошуку замін активів у межах кластера, розуміння структури ринку.

Feature Engineering для кластеризації

import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler

def create_behavioral_features(prices_dict, lookback_days=90):
    features = {}
    
    for symbol, price_series in prices_dict.items():
        returns = price_series.pct_change().dropna()
        
        if len(returns) < lookback_days * 24:  # годинні дані
            continue
        
        recent_returns = returns.iloc[-lookback_days*24:]
        
        features[symbol] = {
            # Характеристики прибутків
            'annualized_return': recent_returns.mean() * 365 * 24,
            'annualized_vol': recent_returns.std() * np.sqrt(365 * 24),
            'sharpe': recent_returns.mean() / (recent_returns.std() + 1e-8) * np.sqrt(365*24),
            
            # Форма розподілу
            'skewness': recent_returns.skew(),
            'kurtosis': recent_returns.kurt(),
            
            # Ризик хвоста
            'var_95': np.percentile(recent_returns, 5),
            'cvar_95': recent_returns[recent_returns <= np.percentile(recent_returns, 5)].mean(),
            
            # Характеристики тренду
            'momentum_30d': price_series.iloc[-720:].pct_change(720).iloc[-1],  # 30-денний прибуток
            'trend_strength': abs(recent_returns.mean()) / (recent_returns.std() + 1e-8),
            
            # Просадка
            'max_drawdown': calculate_max_drawdown(price_series.iloc[-lookback_days*24:]),
            
            # Кореляція з BTC
            'btc_corr': recent_returns.corr(prices_dict.get('BTC', pd.Series()).pct_change().dropna()),
            
            # На основі обсягу
            'avg_daily_volume_usd': get_avg_daily_volume(symbol),
        }
    
    return pd.DataFrame(features).T

Алгоритми кластеризації

K-Means — класичний, швидкий, припускає сферичні кластери:

from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler

def kmeans_clustering(features_df, n_clusters=6, seed=42):
    # Нормалізація
    scaler = StandardScaler()
    features_scaled = scaler.fit_transform(features_df.fillna(0))
    
    # Оптимальне k через метод ліктя
    inertias = []
    k_range = range(2, 15)
    for k in k_range:
        km = KMeans(n_clusters=k, random_state=seed, n_init=10)
        km.fit(features_scaled)
        inertias.append(km.inertia_)
    
    # Вибрати k на "ліктю"
    best_k = find_elbow(inertias, k_range)
    
    km = KMeans(n_clusters=best_k, random_state=seed, n_init=10)
    labels = km.fit_predict(features_scaled)
    
    return labels, km, scaler

DBSCAN — знаходить кластери довільної форми та викиди:

from sklearn.cluster import DBSCAN

def dbscan_clustering(features_scaled, eps=0.5, min_samples=3):
    db = DBSCAN(eps=eps, min_samples=min_samples, metric='euclidean')
    labels = db.fit_predict(features_scaled)
    n_clusters = len(set(labels)) - (1 if -1 in labels else 0)
    n_noise = (labels == -1).sum()
    return labels, n_clusters, n_noise

Ієрархічна кластеризація з дендрограмою:

from scipy.cluster.hierarchy import linkage, dendrogram, fcluster

def hierarchical_clustering(features_scaled, symbols, method='ward', n_clusters=6):
    Z = linkage(features_scaled, method=method)
    
    # Візуалізація дендрограми
    fig, ax = plt.subplots(figsize=(16, 8))
    dendrogram(Z, labels=symbols, orientation='top', ax=ax)
    plt.tight_layout()
    
    labels = fcluster(Z, t=n_clusters, criterion='maxclust')
    return labels, Z

Зменшення розмірності для візуалізації

from sklearn.decomposition import PCA
from sklearn.manifold import TSNE
import umap

def reduce_dimensions(features_scaled, method='umap', n_components=2):
    if method == 'pca':
        reducer = PCA(n_components=n_components, random_state=42)
    elif method == 'tsne':
        reducer = TSNE(n_components=n_components, random_state=42)
    elif method == 'umap':
        reducer = umap.UMAP(n_components=n_components, random_state=42)
    
    embedding = reducer.fit_transform(features_scaled)
    return embedding

Інтерпретація кластерів

Після кластеризації аналізуйте характеристики кожного кластера:

def describe_clusters(features_df, labels):
    features_df['cluster'] = labels
    
    cluster_stats = features_df.groupby('cluster').agg({
        'annualized_return': 'mean',
        'annualized_vol': 'mean',
        'sharpe': 'mean',
        'btc_corr': 'mean',
        'max_drawdown': 'mean',
        'skewness': 'mean'
    }).round(3)
    
    # Назвіть кластери за характеристиками
    cluster_names = {}
    for cluster_id, row in cluster_stats.iterrows():
        if row['btc_corr'] > 0.85 and row['annualized_vol'] > 1.5:
            name = 'High-beta altcoins'
        elif row['btc_corr'] > 0.8 and row['annualized_vol'] < 1.0:
            name = 'Blue-chip crypto'
        elif row['btc_corr'] < 0.5:
            name = 'Decorrelated assets'
        elif row['sharpe'] > 2.0:
            name = 'Strong performers'
        else:
            name = f'Cluster {cluster_id}'
        cluster_names[cluster_id] = name
    
    return cluster_stats, cluster_names

Практичне застосування

Портфельна диверсифікація: вибрати 1–2 активи з кожного кластера для максимальної диверсифікації.

Rotation стратегії: купити найкращий актив з топ-кластера за momentum, ротувати щомісячно.

Аналіз групи аналогів: коли один актив кластера стрімко зростає, знайти відставання в тому ж кластері — потенційні наступні рухи.

Розробка системи кластеризації з методами K-Means/DBSCAN/Hierarchical, UMAP візуалізацією, автоматичною інтерпретацією кластерів та щомісячною ребалансуванням.