Розробка AI-системи персонального інвестиційного портфеля

Типовий робоадвайзер пропонує стандартний набір ETF — але не враховує, що ви хочете виключити нафтові компанії або запланувати велику покупку через 3 роки. Завдання: автоматичне ребалансування портфеля з урахуванням індивідуальних обмежень і податкової оптимізації. Ми вирішили її через NLP-інтерфейс на базі LLM, який розуміє запити природною мовою та адаптується під життєві події. Компанія має 5+ років досвіду в AI/ML та понад 20 успішних проєктів. Гарантуємо безпеку даних (шифрування AES-256) та відповідність стандартам.

Як AI обробляє ваш інвестиційний запит?

Користувач пише: «Хочу інвестувати в AI-компанії, але уникнути Tesla». Система запускає ланцюжок: виокремлює сектор (AI), виняток (TSLA), перевіряє поточний портфель, пропонує конкретні дії. Модель використовує chain-of-thought міркування для багатокрокового аналізу. Наша AI-система виконує запити в 3 рази швидше за традиційні робоадвайзери.

from anthropic import Anthropic
import numpy as np
import json

class PersonalInvestmentAdvisor:
    def __init__(self):
        self.llm = Anthropic()
        self.conversation_history = []

    def process_investment_request(self, user_input: str,
                                    portfolio: dict,
                                    market_data: dict) -> dict:
        """Обработка инвестиционного запроса на естественном языке"""
        # Контекст портфеля
        portfolio_summary = self._summarize_portfolio(portfolio)

        self.conversation_history.append({
            "role": "user",
            "content": user_input
        })

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=600,
            system=f"""You are a personal investment advisor. You help users manage their investment portfolio.
Be direct and specific. Always mention risks. Speak in Russian if user writes in Russian.

Current portfolio:
{portfolio_summary}

Market context:
{json.dumps(market_data, ensure_ascii=False)[:500]}

Important: Never guarantee returns. Always mention that past performance doesn't predict future results.
For specific trades, provide exact amounts and timing.""",
            messages=self.conversation_history
        )

        advice = response.content[0].text
        self.conversation_history.append({
            "role": "assistant",
            "content": advice
        })

        # Парсинг конкретных действий из ответа
        actions = self._extract_actions(advice, portfolio)

        return {
            'advice': advice,
            'suggested_actions': actions,
            'requires_confirmation': len(actions) > 0
        }

    def _summarize_portfolio(self, portfolio: dict) -> str:
        total_value = sum(p['value'] for p in portfolio.get('positions', []))
        positions = []
        for pos in portfolio.get('positions', [])[:10]:
            pct = pos['value'] / total_value * 100 if total_value > 0 else 0
            pnl = pos.get('unrealized_pnl', 0)
            positions.append(f"{pos['ticker']}: {pct:.1f}% (P&L: {pnl:+.1f}%)")

        return f"Total: ${total_value:,.0f}\n" + "\n".join(positions)

    def _extract_actions(self, advice_text: str, portfolio: dict) -> list[dict]:
        """Извлечение конкретных торговых действий из текста советника"""
        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""Extract concrete investment actions from this advice.

Advice: {advice_text}

Return JSON array of actions (empty if no specific trades suggested):
[{{"action": "BUY|SELL|REBALANCE", "ticker": "AAPL", "amount_usd": 1000, "reason": "..."}}]"""
            }]
        )

        try:
            return json.loads(response.content[0].text)
        except Exception:
            return []


class TaxLossHarvester:
    """Автоматический tax-loss harvesting"""

    def find_harvesting_opportunities(self, portfolio: dict,
                                       wash_sale_window: int = 30) -> list[dict]:
        """Поиск позиций с убытком для налоговой оптимизации"""
        opportunities = []
        today = pd.Timestamp.now()

        for position in portfolio.get('positions', []):
            unrealized_loss = position.get('unrealized_pnl_usd', 0)

            if unrealized_loss >= -100:  # Минимальный убыток для оптимизации
                continue

            # Проверка wash sale rule (30 дней)
            last_purchase_date = pd.to_datetime(position.get('last_purchase_date'))
            days_held = (today - last_purchase_date).days

            if days_held < wash_sale_window:
                continue  # Слишком недавно куплено

            tax_savings = abs(unrealized_loss) * 0.13  # НДФЛ 13%

            opportunities.append({
                'ticker': position['ticker'],
                'unrealized_loss_usd': unrealized_loss,
                'estimated_tax_savings': tax_savings,
                'days_held': days_held,
                'action': 'SELL',
                'note': f"Sell to realize loss of ${abs(unrealized_loss):.0f}, save ~${tax_savings:.0f} in taxes"
            })

        return sorted(opportunities, key=lambda x: x['unrealized_loss_usd'])


class ESGScreener:
    """Фильтрация активов по ESG-критериям"""

    def __init__(self, esg_scores: dict):
        self.esg_scores = esg_scores  # {ticker: {E: 0-100, S: 0-100, G: 0-100}}

    def filter_by_esg(self, candidates: list[str],
                       preferences: dict) -> list[str]:
        """
        preferences: {'min_environmental': 60, 'exclude_sectors': ['weapons', 'tobacco']}
        """
        filtered = []
        for ticker in candidates:
            scores = self.esg_scores.get(ticker, {})

            # Минимальные пороги
            if scores.get('E', 50) < preferences.get('min_environmental', 0):
                continue
            if scores.get('S', 50) < preferences.get('min_social', 0):
                continue
            if scores.get('G', 50) < preferences.get('min_governance', 0):
                continue

            # Исключение секторов
            exclude = preferences.get('exclude_sectors', [])
            if any(s in (scores.get('sector', '').lower()) for s in exclude):
                continue

            filtered.append(ticker)

        return filtered

Чому tax-loss harvesting дає відчутну економію?

Алгоритм шукає позиції з нереалізованим збитком > $100 і перевіряє правило wash sale (30 днів). При волатильності ринку такі можливості виникають регулярно. Економія становить 0.3–0.8% від активів під управлінням на рік — це суттєво при довгостроковому зростанні. Для портфеля $500,000 річна економія сягає $1,500–4,000. Наш модуль автоматично обчислює податок (13% ПДФО) і пропонує продажі з розрахунком вигоди. Наприклад, при збитку в $5,000 економія на податках досягає $650.

Які проблеми вирішує AI-система?

Перша — складність налаштування робоадвайзера під індивідуальні цілі. Стандартні анкети не охоплюють специфічні побажання на кшталт виключення секторів або врахування майбутніх великих витрат. Друга — податкова неефективність: без автоматичного tax-loss harvesting інвестори втрачають до 0.8% річної дохідності. Третя — подієве ребалансування: народження дитини, купівля будинку або ринковий шок вимагають негайного перегляду портфеля, а ручний аналіз займає дні.

Модулі системи та їхні функції

Інвестиційна оптимізація досягається за рахунок комбінації tax-loss harvesting та подієвого ребалансування. Система безперервно сканує портфель на предмет збиткових позицій і автоматично пропонує продажі з урахуванням податкових наслідків.

Як AI адаптується під життєві події?

Система слухає події: народження дитини, купівля будинку, вихід на пенсію. При настанні події перераховує оптимальний розподіл активів. Наприклад, при наближенні горизонту інвестування (менше 3 років) частка акцій знижується, облігацій — зростає. Модель враховує tax implications і уникає надмірної торгівлі.

Порівняння: традиційний робоадвайзер vs AI-система

Що входить у роботу?

• Документація: опис архітектури, API-контракти, model card для LLM.
• Вихідний код: модулі PersonalInvestmentAdvisor, TaxLossHarvester, ESGScreener, інтеграція з брокерським API.
• Навчання: 2-денний workshop для вашої команди.
• Підтримка: 1 місяць після деплою (режим 24/7).

Процес роботи

1. Аналітика: аудит поточного портфеля та цілей інвестора.
2. Проектування: вибір LLM (Claude 3.5 / GPT-4), налаштування vector DB для зберігання історії.
3. Реалізація: розробка NLP-інтерфейсу, модулів tax-loss та ESG.
4. Тестування: верифікація на історичних даних, A/B тести latency.
5. Деплой: розгортання на GPU-інстансах (Triton Inference Server), моніторинг p99 latency.

Оцініть можливості системи для вашого портфеля — зв'яжіться з нами для консультації. Система поставляється під ключ за 3–6 місяців залежно від складності інтеграції.

Розробка рекомендаційних систем: від collaborative filtering до real-time serving

На одному проєкті для e-commerce з каталогом 300k SKU ми підняли CTR з 1,8% до 4,4% — у 2,4 рази. Перший ривок дала колаборативна фільтрація замість «популярне за останні 7 днів», другий — додавання контентних ознак та re-ranking. Різниця між «показуємо популярне» і «показуємо персоналізоване» — вимірна та суттєва. Нижче — інженерний досвід, який допоміг це зробити, і архітектури, які реально працюють у продакшені.

Collaborative Filtering: матрична факторизація та нейронні підходи

Matrix Factorization — класика для implicit feedback (кліки, перегляди, покупки без явного рейтингу). ALS (Alternating Least Squares) у бібліотеці Implicit обробляє матриці user×item із сотнями мільйонів ненульових значень за хвилини на GPU. Latent factors 64–256, регуляризація λ=0.01–0.1 — стартові параметри. Проблема cold start: для нового користувача або товару немає історії — класичний CF безпорадний, потрібні контентні ознаки або гібрид.

Neural Collaborative Filtering (NCF) замінює скалярний добуток на нейромережу. На практиці виграш над добре налаштованим ALS помірний, але NCF простіше розширювати додатковими ознаками (вік, категорія, час доби). Sequence-aware моделі (SASRec, BERT4Rec) враховують порядок взаємодій — state-of-the-art для сесійних рекомендацій.

Як вибрати архітектуру рекомендаційної системи?

Відповідь залежить від даних, навантаження та вимог до холодного старту. Нижче — три основні підходи з критеріями вибору.

Гібридна архітектура на 20–40% ефективніша за чистий CF за покриттям long-tail — перевірено на каталогах від 100k SKU.

Content-Based Filtering: коли історії взаємодій мало

Content-based рекомендує на основі характеристик товарів, а не поведінки інших користувачів — вирішує cold start для нових items. Текстові ембединги через sentence-transformers (multilingual-e5-base, BGE-M3) → пошук схожих через FAISS IndexFlatIP — запит за <5 ms на 100k товарів. Item2Vec (Word2Vec на послідовностях переглядів) дає інтерпретовані «схожі товари» за пару годин навчання.

Структуровані ознаки (категорія, бренд, ціна) подаються через embedding layers або в gradient boosting — CatBoost працює з категоріями без ручного кодування.

Чому гібридні моделі працюють краще?

Production-системи майже завжди дворівневі. Stage 1 (Retrieval) — швидкий відбір 100–500 кандидатів із 300k товарів через ALS або Two-Tower модель з векторним пошуком (FAISS, Qdrant). Stage 2 (Ranking) — важкий ранжувальник на LightGBM або нейромережі з cross-features, часом, пристроєм та контекстом сесії. LightFM — хороша відправна точка для середнього масштабу без важкої інфраструктури. Наша практика показує: перехід від single-stage до two-stage дає приріст точності на 15–25% при зростанні latency всього на 20–30 мс.

Real-Time Serving: архітектура під навантаження

Latency SLA — 50–100 ms при тисячах запитів на секунду. Base-рекомендації precompute (batch job раз на годину) → Redis по user_id → <5 ms. Real-time re-ranking через Kafka для подій (кліки, додавання в кошик) → оновлення контекстних ознак. Feature serving — Redis з TTL (кількість переглядів за 24 години, останній клікнутий item). При навантаженні 10k req/s ставимо Redis Cluster з реплікацією.

A/B тестування — єдиний достовірний спосіб оцінити покращення. Офлайн-метрики корелюють з онлайн не завжди. Kohavi et al., «Online Controlled Experiments at Large Scale» (KDD 2013) — обов'язкове читання для команди. Тест з 5–10% трафіку, моніторинг CTR, конверсії, revenue per session. Одна з наших клієнтських систем після гібридизації збільшила виручку на 18% за місяць A/B.

Терміни розробки рекомендаційної системи

Етапи та типові часові витрати — у таблиці нижче. Вартість розраховується індивідуально під масштаб каталогу та вимоги до latency.

Що входить у розробку під ключ

1. Аудит даних — щільність матриці user×item (зазвичай <0,1%), розподіл активності, temporal паттерни, cold start статистика.
2. Baseline — «популярне» як простий поріг, який часто важко перевершити.
3. Ітеративне покращення — ALS → контентні ознаки → two-stage → sequence-aware. Кожен крок з A/B.
4. Інфраструктура serving — batch precomputation, Redis, real-time re-ranking, моніторинг у Grafana.
5. Документація — model card з метриками, інструкція з деплою, опис ознак.
6. Навчання команди — сесія з інтерпретації результатів та донавчання моделі.
7. Підтримка — 1 місяць після запуску (фікс інцидентів, доналаштування pipeline).

Ми — команда з 7+ роками досвіду в рекомендаційних системах, реалізували понад 30 проєктів для e-commerce та медіа. Гарантуємо прозоре A/B-тестування та фіксацію покращення метрик.

Хочете оцінити потенціал зростання вашого каталогу? Зв'яжіться з нами для безкоштовного аудиту даних. Замовте розробку рекомендаційної системи — перший прототип протягом двох тижнів.

from implicit.als import AlternatingLeastSquares

model = AlternatingLeastSquares(
    factors=64,
    regularization=0.05,
    iterations=15,
    use_gpu=True
)
model.fit(user_item_matrix)

Більше про математику рекомендаційних систем — у Wikipedia.