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

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

Персональная инвестиционная система идёт глубже робоадвайзера: учитывает индивидуальные ценности (ESG инвестирование), налоговую оптимизацию через tax-loss harvesting, ситуативные события (рождение ребёнка → ребалансировка под образование), и естественноязыковые запросы типа "хочу инвестировать в AI-компании но избежать Tesla".

Персонализация через естественный язык

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 даёт экономию 0.3-0.8% AuM в год при волатильном рынке. ESG-фильтрация снижает инвестиционную вселенную на 20-30%, что незначительно влияет на доходность при правильной диверсификации (исследования показывают 0-0.5% penalty).