""" Strategy Reporter — LLM-generated weekly strategy letters ========================================================== Produces human-readable analysis of allocation decisions, market conditions, and risk assessments. Designed for both Telegram delivery and on-chain attestation via ERC-8004. """ import json import logging import os import time from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Dict, List, Optional logger = logging.getLogger("strategy_reporter") @dataclass class StrategyReport: """Weekly strategy report with allocation rationale.""" timestamp: float report_id: str period: str # e.g., "2026-W17" # Portfolio state current_weights: Dict[str, float] portfolio_value_usd: float period_return_pct: float # Market conditions market_summary: str yield_environment: str risk_assessment: str # Allocation rationale allocation_rationale: str key_trades: List[str] # Forward-looking outlook: str risk_factors: List[str] # Full narrative full_report: str # On-chain hash (for ERC-8004 attestation) content_hash: Optional[str] = None def to_dict(self) -> Dict[str, Any]: return { "timestamp": self.timestamp, "report_id": self.report_id, "period": self.period, "weights": self.current_weights, "portfolio_value": self.portfolio_value_usd, "period_return": self.period_return_pct, "market_summary": self.market_summary, "yield_environment": self.yield_environment, "risk_assessment": self.risk_assessment, "allocation_rationale": self.allocation_rationale, "key_trades": self.key_trades, "outlook": self.outlook, "risk_factors": self.risk_factors, "full_report": self.full_report, "content_hash": self.content_hash, } def to_telegram_message(self) -> str: """Format report for Telegram delivery.""" weights_str = " | ".join( f"{k}: {v*100:.1f}%" for k, v in self.current_weights.items() ) trades_str = "\n".join(f" • {t}" for t in self.key_trades) if self.key_trades else " • No trades this period" risks_str = "\n".join(f" ⚠️ {r}" for r in self.risk_factors) return f"""📊 **RWA Yield Router — Weekly Report** ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 📅 Period: {self.period} 💰 Portfolio Value: ${self.portfolio_value_usd:,.2f} 📈 Period Return: {self.period_return_pct:+.2f}% 📊 **Current Allocation** {weights_str} 🔄 **Key Trades** {trades_str} 📈 **Market Summary** {self.market_summary} 💡 **Yield Environment** {self.yield_environment} 🛡️ **Risk Assessment** {self.risk_assessment} 🎯 **Allocation Rationale** {self.allocation_rationale} 🔮 **Outlook** {self.outlook} ⚠️ **Risk Factors** {risks_str} ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 🤖 Generated by Dynamic RWA Yield Router """ class StrategyReporter: """ Generates strategy reports using LLM or rule-based templates. Supports: - OpenAI GPT-4 / Anthropic Claude API integration - Fallback rule-based template system - On-chain content hashing for ERC-8004 attestation """ def __init__( self, use_llm: bool = True, llm_api_key: Optional[str] = None, llm_model: str = "gpt-4o", ): self.use_llm = use_llm self.llm_api_key = llm_api_key or os.getenv("OPENAI_API_KEY") self.llm_model = llm_model self.report_history: List[StrategyReport] = [] def generate_report( self, current_weights: Dict[str, float], portfolio_value: float, period_return: float, yield_data: Dict[str, float], risk_summary: Dict, execution_history: List[Dict], macro_data: Optional[Dict] = None, ) -> StrategyReport: """ Generate a weekly strategy report. Tries LLM first, falls back to template-based generation. """ now = datetime.now(timezone.utc) period = now.strftime("%Y-W%V") report_id = f"report_{int(time.time())}" # Gather context context = { "weights": current_weights, "portfolio_value": portfolio_value, "period_return": period_return, "yields": yield_data, "risk": risk_summary, "trades": execution_history[-5:] if execution_history else [], "macro": macro_data or {}, "timestamp": now.isoformat(), } if self.use_llm and self.llm_api_key: try: return self._generate_llm_report(context, period, report_id) except Exception as e: logger.warning(f"LLM report generation failed: {e}, using template") return self._generate_template_report(context, period, report_id) def _generate_template_report( self, context: Dict, period: str, report_id: str ) -> StrategyReport: """Rule-based template report generation.""" weights = context["weights"] yields = context["yields"] risk = context["risk"] # Market summary usdy_yield = yields.get("usdy", 4.25) meth_yield = yields.get("meth", 3.5) mi4_yield = yields.get("mi4", 5.0) if usdy_yield > 5.0: yield_env = "elevated" market_tone = "Risk-free yields remain attractive as T-bill rates hold above 5%." elif usdy_yield > 4.0: yield_env = "moderate" market_tone = "Treasury yields provide a solid baseline at current levels." else: yield_env = "compressed" market_tone = "Declining risk-free rates push capital toward higher-yielding alternatives." # Determine dominant allocation max_asset = max(weights, key=weights.get) max_weight = weights[max_asset] if max_asset == "USDY" and max_weight > 0.45: allocation_rationale = ( f"Overweight USDY ({max_weight*100:.0f}%) reflects a defensive stance. " f"At {usdy_yield:.2f}% APY, tokenized T-bills offer compelling risk-adjusted " f"returns versus volatile crypto-native yields. The agent prioritizes " f"capital preservation in the current risk environment." ) elif max_asset == "mETH" and max_weight > 0.40: allocation_rationale = ( f"Elevated mETH allocation ({max_weight*100:.0f}%) captures staking yield " f"plus ETH price appreciation potential. At {meth_yield:.2f}% APY with " f"Mantle's liquid staking infrastructure, this represents the best " f"risk-reward in the current market." ) else: allocation_rationale = ( f"Balanced allocation across USDY ({weights.get('USDY', 0)*100:.0f}%), " f"mETH ({weights.get('mETH', 0)*100:.0f}%), and MI4 ({weights.get('MI4', 0)*100:.0f}%) " f"optimizes for diversification. The RL agent has identified this mix as " f"maximizing risk-adjusted yield given current correlations." ) # Key trades key_trades = [] for trade in context.get("trades", []): if isinstance(trade, dict): summary = trade.get("human_summary", "") if summary: key_trades.append(summary[:100]) if not key_trades: key_trades = ["No significant rebalancing this period"] # Risk factors risk_factors = [] risk_level = risk.get("latest_risk_level", "low") if risk_level in ("high", "critical"): risk_factors.append(f"Overall risk level: {risk_level.upper()}") drawdown = risk.get("current_drawdown", 0) if drawdown > 0.03: risk_factors.append(f"Portfolio drawdown at {drawdown*100:.1f}%") risk_factors.extend([ "Smart contract risk: DeFi protocol exploits remain a tail risk", "Regulatory risk: RWA tokenization regulatory clarity evolving", "Liquidity risk: DEX depth for RWA pairs may thin in stress", ]) # Outlook if context.get("period_return", 0) > 0: outlook = ( f"Positive momentum continues with {context['period_return']:.2f}% returns. " f"The agent maintains its current strategy, monitoring yield curve dynamics " f"and mETH staking demand for next adjustment signals." ) else: outlook = ( f"Defensive positioning warranted after {context['period_return']:.2f}% drawdown. " f"Shifting toward higher USDY allocation while monitoring recovery signals. " f"Key catalysts: Fed rate decision, ETH staking rate changes, MI4 rebalance." ) # Full narrative full_report = f"""# Dynamic RWA Yield Router — Strategy Report ## Period: {period} ### Executive Summary Portfolio value: ${context['portfolio_value']:,.2f} | Period return: {context['period_return']:+.2f}% Current allocation: {', '.join(f'{k}={v*100:.0f}%' for k, v in weights.items())} ### Market Environment {market_tone} The yield landscape is {yield_env}: - USDY (tokenized T-bills): {usdy_yield:.2f}% APY - mETH (Mantle staked ETH): {meth_yield:.2f}% APY - MI4 (tokenized index): {mi4_yield:.2f}% APY ### Allocation Rationale {allocation_rationale} ### Risk Assessment Overall risk: {risk_level.upper()} (score: {risk.get('latest_risk_score', 0):.3f}) Circuit breaker: {risk.get('circuit_breaker', 'closed')} {chr(10).join(f'- {r}' for r in risk_factors)} ### Outlook {outlook} --- *Generated by Dynamic RWA Yield Router v1.0 — an autonomous AI agent on Mantle L2* *Report ID: {report_id}* """ import hashlib content_hash = "0x" + hashlib.sha256(full_report.encode()).hexdigest() report = StrategyReport( timestamp=time.time(), report_id=report_id, period=period, current_weights=weights, portfolio_value_usd=context["portfolio_value"], period_return_pct=context["period_return"], market_summary=market_tone, yield_environment=f"Yield environment is {yield_env}. USDY: {usdy_yield:.2f}%, mETH: {meth_yield:.2f}%, MI4: {mi4_yield:.2f}%", risk_assessment=f"Risk level: {risk_level.upper()}. Score: {risk.get('latest_risk_score', 0):.3f}", allocation_rationale=allocation_rationale, key_trades=key_trades, outlook=outlook, risk_factors=risk_factors, full_report=full_report, content_hash=content_hash, ) self.report_history.append(report) return report def _generate_llm_report( self, context: Dict, period: str, report_id: str ) -> StrategyReport: """Generate report via OpenAI API.""" import openai client = openai.OpenAI(api_key=self.llm_api_key) system_prompt = """You are an AI portfolio analyst for the Dynamic RWA Yield Router, an autonomous agent managing a portfolio of Real World Asset tokens on Mantle L2. Write a concise, professional weekly strategy letter covering: 1. Market environment and yield landscape 2. Current allocation rationale (data-driven) 3. Key trades executed and why 4. Risk assessment 5. Forward-looking outlook Style: Bloomberg terminal meets crypto-native. Precise, quantitative, actionable. Keep it under 800 words.""" user_prompt = f"""Generate the weekly strategy report for period {period}. Portfolio State: {json.dumps(context, indent=2, default=str)} Focus on explaining WHY the RL agent chose these specific weights given the yield environment and risk conditions.""" response = client.chat.completions.create( model=self.llm_model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=0.7, max_tokens=2000, ) full_report = response.choices[0].message.content import hashlib content_hash = "0x" + hashlib.sha256(full_report.encode()).hexdigest() report = StrategyReport( timestamp=time.time(), report_id=report_id, period=period, current_weights=context["weights"], portfolio_value_usd=context["portfolio_value"], period_return_pct=context["period_return"], market_summary="See full report", yield_environment="See full report", risk_assessment="See full report", allocation_rationale="See full report", key_trades=["See full report"], outlook="See full report", risk_factors=["See full report"], full_report=full_report, content_hash=content_hash, ) self.report_history.append(report) return report