Разработка рекомендательной системы крипто-активов

Разработка рекомендательной системы крипто-активов

Рекомендательная система для крипто-активов — это не Netflix «вам понравится этот фильм». Это система, которая на основе профиля пользователя (риск-аппетит, горизонт инвестирования, текущий портфель) и текущих рыночных данных предлагает активы для покупки, продажи или ребалансировки.

Постановка задачи

Контент-based рекомендации: рекомендуем активы похожие на те, которые пользователь уже держит или которые понравились.

Collaborative filtering: рекомендуем активы, которые держат похожие пользователи (по профилю риска, размеру портфеля, горизонту).

Hybrid approach: комбинация обоих методов + market signals.

Context-aware: учитываем текущие рыночные условия — в медвежьем рынке другие рекомендации, чем в бычьем.

User profiling

from dataclasses import dataclass
from enum import Enum

class RiskProfile(Enum):
    CONSERVATIVE = 'conservative'
    MODERATE = 'moderate'
    AGGRESSIVE = 'aggressive'
    DEGEN = 'degen'

@dataclass
class UserProfile:
    user_id: str
    risk_profile: RiskProfile
    investment_horizon_days: int     # 30, 90, 365, 1095...
    portfolio_value_usd: float
    current_holdings: dict           # {symbol: value_usd}
    preferred_categories: list       # ['defi', 'layer1', 'gaming']
    excluded_categories: list        # ['meme', 'high_leverage']
    max_single_asset_pct: float      # 0.20 = max 20% в одном активе
    
    def get_available_budget(self):
        """Сколько не задействовано"""
        invested = sum(self.current_holdings.values())
        return max(0, self.portfolio_value_usd - invested)

Asset scoring engine

import numpy as np
import pandas as pd
from typing import List, Dict

class AssetScorer:
    def __init__(self, market_data, on_chain_data=None):
        self.market_data = market_data
        self.on_chain_data = on_chain_data
    
    def score_asset(self, symbol, user_profile, market_regime):
        """Финальный score = взвешенная сумма sub-scores"""
        scores = {}
        weights = self._get_weights_for_profile(user_profile.risk_profile)
        
        # 1. Momentum score
        scores['momentum'] = self._momentum_score(symbol)
        
        # 2. Risk-adjusted return score
        scores['risk_return'] = self._risk_adjusted_score(symbol, user_profile)
        
        # 3. Diversification benefit score
        scores['diversification'] = self._diversification_score(
            symbol, user_profile.current_holdings
        )
        
        # 4. Liquidity score
        scores['liquidity'] = self._liquidity_score(symbol, user_profile.portfolio_value_usd)
        
        # 5. Sentiment score
        scores['sentiment'] = self._sentiment_score(symbol)
        
        # 6. Fundamentals score (on-chain + protocol metrics)
        scores['fundamentals'] = self._fundamentals_score(symbol)
        
        # 7. Market regime fit
        scores['regime_fit'] = self._regime_fit_score(symbol, market_regime)
        
        # Взвешенный итоговый score
        final_score = sum(scores[k] * weights.get(k, 0.1) for k in scores)
        
        return {
            'symbol': symbol,
            'final_score': final_score,
            'sub_scores': scores,
            'weights': weights
        }
    
    def _momentum_score(self, symbol):
        """Нормализованный momentum score"""
        for period_days, weight in [(7, 0.3), (30, 0.5), (90, 0.2)]:
            period_hours = period_days * 24
            prices = self.market_data[symbol]['close']
            return_period = prices.iloc[-1] / prices.iloc[-period_hours] - 1
            # z-score относительно вселенной активов
        return np.clip(return_period / 0.5, -1, 1)  # упрощённо
    
    def _diversification_score(self, symbol, current_holdings):
        """Насколько актив диверсифицирует существующий портфель"""
        if not current_holdings:
            return 0.5  # нейтральный если портфель пустой
        
        symbol_returns = self.market_data[symbol]['close'].pct_change().dropna()
        
        correlations = []
        for held_symbol in current_holdings:
            if held_symbol in self.market_data:
                held_returns = self.market_data[held_symbol]['close'].pct_change().dropna()
                corr = symbol_returns.corr(held_returns)
                correlations.append(abs(corr))
        
        if not correlations:
            return 0.5
        
        avg_correlation = np.mean(correlations)
        # Низкая корреляция = высокий diversification score
        return 1 - avg_correlation
    
    def _get_weights_for_profile(self, risk_profile):
        WEIGHTS = {
            RiskProfile.CONSERVATIVE: {
                'momentum': 0.10, 'risk_return': 0.30, 'diversification': 0.20,
                'liquidity': 0.20, 'sentiment': 0.05, 'fundamentals': 0.10,
                'regime_fit': 0.05
            },
            RiskProfile.MODERATE: {
                'momentum': 0.20, 'risk_return': 0.20, 'diversification': 0.15,
                'liquidity': 0.15, 'sentiment': 0.10, 'fundamentals': 0.10,
                'regime_fit': 0.10
            },
            RiskProfile.AGGRESSIVE: {
                'momentum': 0.35, 'risk_return': 0.10, 'diversification': 0.05,
                'liquidity': 0.10, 'sentiment': 0.20, 'fundamentals': 0.05,
                'regime_fit': 0.15
            }
        }
        return WEIGHTS.get(risk_profile, WEIGHTS[RiskProfile.MODERATE])

Portfolio construction из рекомендаций

def build_recommended_portfolio(user_profile, scored_assets, max_positions=10):
    """
    Из scored assets строим оптимальный портфель
    """
    # Фильтрация
    filtered = [a for a in scored_assets 
                if a['symbol'] not in user_profile.excluded_categories
                and a['final_score'] > 0.3]
    
    # Сортировка по score
    filtered.sort(key=lambda x: x['final_score'], reverse=True)
    
    # Отбор топ-N с учётом диверсификации
    selected = []
    selected_categories = set()
    
    for asset in filtered[:50]:  # смотрим топ-50
        if len(selected) >= max_positions:
            break
        
        # Не берём больше 3 из одной категории
        category = get_asset_category(asset['symbol'])
        if selected_categories.count(category) >= 3:
            continue
        
        selected.append(asset)
        selected_categories.add(category)
    
    # Allocation (взвешивание по score)
    total_score = sum(a['final_score'] for a in selected)
    available_budget = user_profile.get_available_budget()
    
    allocations = []
    for asset in selected:
        raw_allocation = (asset['final_score'] / total_score) * available_budget
        
        # Ограничиваем максимальную долю
        max_alloc = user_profile.portfolio_value_usd * user_profile.max_single_asset_pct
        final_allocation = min(raw_allocation, max_alloc)
        
        allocations.append({
            'symbol': asset['symbol'],
            'allocation_usd': final_allocation,
            'allocation_pct': final_allocation / user_profile.portfolio_value_usd,
            'score': asset['final_score'],
            'sub_scores': asset['sub_scores']
        })
    
    return allocations

Explainability — почему рекомендован этот актив

Пользователю важно понимать логику рекомендации:

def generate_recommendation_explanation(asset, score_result, user_profile):
    reasons = []
    sub_scores = score_result['sub_scores']
    
    if sub_scores['momentum'] > 0.7:
        reasons.append(f"Сильный momentum: +{get_return(asset, 30):.1f}% за 30 дней")
    
    if sub_scores['diversification'] > 0.7:
        reasons.append(f"Низкая корреляция с вашим портфелем ({get_correlation(asset):.2f})")
    
    if sub_scores['fundamentals'] > 0.7:
        reasons.append(f"Высокая on-chain активность: TVL вырос на {get_tvl_change(asset):.0f}%")
    
    if sub_scores['sentiment'] > 0.6:
        reasons.append("Позитивный sentiment в сообществе за последние 7 дней")
    
    risk_warning = []
    if get_volatility(asset) > 0.8:
        risk_warning.append("Высокая волатильность")
    if get_market_cap(asset) < 100_000_000:
        risk_warning.append("Малая капитализация — высокий риск")
    
    return {
        'reasons': reasons,
        'risk_warnings': risk_warning,
        'confidence': score_result['final_score']
    }

Dashboard и UX

Рекомендательная панель:

• Топ-10 рекомендаций с score breakdown
• Объяснение каждой рекомендации на простом языке
• Кнопка «Применить» для автоматического исполнения через exchange API
• Ожидаемые характеристики рекомендуемого портфеля (risk/return projection)

Персонализация: настройки профиля риска, горизонта, excluded categories.

Performance tracking: насколько прибыльными оказались прошлые рекомендации — backtesting рекомендательного алгоритма.

Разрабатываем рекомендательную систему с multi-factor scoring, user profiling, portfolio construction с diversification constraints и explainable recommendations через React веб-интерфейс.