From 2cc9e7b20be016a12eb2c2304203fdf76dfe2a87 Mon Sep 17 00:00:00 2001 From: Mehran Moazeni Date: Tue, 9 Dec 2025 22:28:00 +0100 Subject: [PATCH 01/11] Add multi-asset allocation agent, CLI entry, and PDF output --- .../agents/__init__.py | 15 + .../agents/multi_asset_allocation_agent.py | 314 ++++++++++++++++++ src/tradegraph_financial_advisor/main.py | 124 ++++++- .../reporting/__init__.py | 3 +- .../reporting/multi_asset_reporter.py | 162 +++++++++ tests/unit/test_multi_asset_agent.py | 57 ++++ 6 files changed, 671 insertions(+), 4 deletions(-) create mode 100644 src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py create mode 100644 src/tradegraph_financial_advisor/reporting/multi_asset_reporter.py create mode 100644 tests/unit/test_multi_asset_agent.py diff --git a/src/tradegraph_financial_advisor/agents/__init__.py b/src/tradegraph_financial_advisor/agents/__init__.py index e69de29..ea22f0c 100644 --- a/src/tradegraph_financial_advisor/agents/__init__.py +++ b/src/tradegraph_financial_advisor/agents/__init__.py @@ -0,0 +1,15 @@ +from .channel_report_agent import ChannelReportAgent +from .financial_agent import FinancialAnalysisAgent +from .news_agent import NewsReaderAgent +from .recommendation_engine import TradingRecommendationEngine +from .report_analysis_agent import ReportAnalysisAgent +from .multi_asset_allocation_agent import MultiAssetAllocationAgent + +__all__ = [ + "ChannelReportAgent", + "FinancialAnalysisAgent", + "NewsReaderAgent", + "TradingRecommendationEngine", + "ReportAnalysisAgent", + "MultiAssetAllocationAgent", +] diff --git a/src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py b/src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py new file mode 100644 index 0000000..af1bb46 --- /dev/null +++ b/src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py @@ -0,0 +1,314 @@ +"""Agent that creates allocation plans across stocks, ETFs, and crypto for various horizons.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Dict, List, Optional + +from loguru import logger + +from .base_agent import BaseAgent + + +@dataclass +class AllocationSuggestion: + asset_class: str + weight: float + rationale: str + sample_assets: List[Dict[str, str]] + + +HORIZON_LABELS = { + "1w": "1-Week", + "1m": "1-Month", + "1y": "1-Year", +} + + +STRATEGY_LIBRARY: Dict[str, Dict[str, Dict[str, AllocationSuggestion]]] = {} + + +def _build_strategy_library() -> Dict[str, Dict[str, Dict[str, AllocationSuggestion]]]: + asset_pool = { + "stocks": [ + {"symbol": "AAPL", "thesis": "Cash-rich mega-cap"}, + {"symbol": "MSFT", "thesis": "Enterprise AI exposure"}, + {"symbol": "NVDA", "thesis": "GPU leadership"}, + {"symbol": "AMZN", "thesis": "Cloud + retail"}, + ], + "etfs": [ + {"symbol": "VOO", "thesis": "S&P 500 core"}, + {"symbol": "QQQ", "thesis": "Large-cap growth"}, + {"symbol": "ARKK", "thesis": "High beta innovation"}, + {"symbol": "TLT", "thesis": "Long-duration bonds"}, + ], + "crypto": [ + {"symbol": "BTC", "thesis": "Digital gold"}, + {"symbol": "ETH", "thesis": "Smart contracts"}, + {"symbol": "SOL", "thesis": "High throughput L1"}, + ], + } + + def suggestion(asset_class: str, weight: float, rationale: str) -> AllocationSuggestion: + return AllocationSuggestion( + asset_class=asset_class, + weight=weight, + rationale=rationale, + sample_assets=asset_pool[asset_class][:2], + ) + + strategies = { + "growth": { + "1w": { + "stocks": suggestion( + "stocks", 0.35, "Stay liquid in quality mega caps while watching catalysts." + ), + "etfs": suggestion( + "etfs", 0.15, "Use QQQ/ARKK for beta exposure without security selection." + ), + "crypto": suggestion( + "crypto", 0.50, "Lean into BTC/ETH momentum for tactical upside." + ), + }, + "1m": { + "stocks": suggestion( + "stocks", 0.45, "Compound AI and cloud tailwinds via mega caps." + ), + "etfs": suggestion( + "etfs", 0.20, "Blend sector ETFs for diversification." + ), + "crypto": suggestion( + "crypto", 0.35, "Maintain crypto beta for asymmetric upside." + ), + }, + "1y": { + "stocks": suggestion( + "stocks", 0.5, "Core equity growth allocation with reinvestment." + ), + "etfs": suggestion( + "etfs", 0.25, "Add thematic ETFs to capture innovation baskets." + ), + "crypto": suggestion( + "crypto", 0.25, "Long-term conviction in BTC/ETH network effects." + ), + }, + }, + "balanced": { + "1w": { + "stocks": suggestion( + "stocks", 0.3, "Blend defensives with growth to dampen volatility." + ), + "etfs": suggestion( + "etfs", 0.4, "VOO/TLT core to keep drawdowns manageable." + ), + "crypto": suggestion( + "crypto", 0.3, "Measured crypto sleeve for opportunistic moves." + ), + }, + "1m": { + "stocks": suggestion( + "stocks", 0.4, "Add cyclicals selectively as macro visibility improves." + ), + "etfs": suggestion( + "etfs", 0.4, "Core passive ETFs to anchor risk." + ), + "crypto": suggestion( + "crypto", 0.2, "Keep crypto beta but size for volatility." + ), + }, + "1y": { + "stocks": suggestion( + "stocks", 0.45, "Dividend growers + quality compounders." + ), + "etfs": suggestion( + "etfs", 0.4, "Broad equity and bond ETFs for balance." + ), + "crypto": suggestion( + "crypto", 0.15, "Smaller crypto sleeve for optionality." + ), + }, + }, + "defensive": { + "1w": { + "stocks": suggestion( + "stocks", 0.25, "Prefer healthcare and staples for stability." + ), + "etfs": suggestion( + "etfs", 0.6, "High-quality bond and minimum-vol ETFs." + ), + "crypto": suggestion( + "crypto", 0.15, "Tiny crypto exposure to stay engaged." + ), + }, + "1m": { + "stocks": suggestion( + "stocks", 0.3, "Income-oriented equities." + ), + "etfs": suggestion( + "etfs", 0.55, "Blend IG bonds with broad equity ETFs." + ), + "crypto": suggestion( + "crypto", 0.15, "Cap risk but allow for upside." + ), + }, + "1y": { + "stocks": suggestion( + "stocks", 0.35, "Quality and value tilt." + ), + "etfs": suggestion( + "etfs", 0.5, "VOO/TLT core plus dividend ETFs." + ), + "crypto": suggestion( + "crypto", 0.15, "Long-dated call option sized exposure." + ), + }, + }, + "income": { + "1w": { + "stocks": suggestion( + "stocks", 0.35, "Dividend aristocrats for short-term distributions." + ), + "etfs": suggestion( + "etfs", 0.55, "Covered-call and bond ETFs for yield." + ), + "crypto": suggestion( + "crypto", 0.10, "Stablecoin yield or staking." + ), + }, + "1m": { + "stocks": suggestion( + "stocks", 0.4, "REITs + utilities blend." + ), + "etfs": suggestion( + "etfs", 0.5, "Bond ladders and dividend ETFs." + ), + "crypto": suggestion( + "crypto", 0.1, "Select staking strategies." + ), + }, + "1y": { + "stocks": suggestion( + "stocks", 0.45, "Global dividend growth." + ), + "etfs": suggestion( + "etfs", 0.45, "Income ETFs and bond funds." + ), + "crypto": suggestion( + "crypto", 0.1, "Yield-focused crypto vehicles." + ), + }, + }, + } + return strategies + + +STRATEGY_LIBRARY = _build_strategy_library() + + +class MultiAssetAllocationAgent(BaseAgent): + def __init__(self, **kwargs): + super().__init__( + name="MultiAssetAllocationAgent", + description="Builds allocations across stocks, ETFs, and crypto for multiple horizons", + **kwargs, + ) + + async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: + budget = float(input_data.get("budget", 0)) + if budget <= 0: + raise ValueError("Budget must be greater than zero.") + strategies = input_data.get("strategies") or ["balanced"] + normalized = self._normalize_strategies(strategies) + logger.info( + "Running multi-asset allocation for budget %.2f with strategies %s", + budget, + normalized, + ) + + plans = [self._build_plan(strategy, budget) for strategy in normalized] + advisory_notes = [ + "Allocations are illustrative; rebalance as macro drivers evolve.", + "Size crypto sleeves according to volatility tolerance and access to custody.", + "ETFs offer quick diversification for both beta and fixed-income exposures.", + ] + + return { + "budget": budget, + "strategies": plans, + "notes": advisory_notes, + } + + def _normalize_strategies(self, strategies: List[str]) -> List[str]: + valid = [] + for strategy in strategies: + key = strategy.lower().strip() + if key in STRATEGY_LIBRARY: + valid.append(key) + if not valid: + valid = ["balanced"] + return valid + + def _build_plan(self, strategy: str, budget: float) -> Dict[str, Any]: + template = STRATEGY_LIBRARY[strategy] + horizons = {} + for horizon, allocations in template.items(): + horizons[horizon] = self._build_horizon_allocations( + horizon, allocations, budget + ) + return { + "strategy": strategy, + "description": self._describe_strategy(strategy), + "horizons": horizons, + } + + def _describe_strategy(self, strategy: str) -> str: + descriptions = { + "growth": "Aggressive mix leaning into innovation, AI, and crypto beta.", + "balanced": "Even-handed mix balancing upside with drawdown control.", + "defensive": "Capital preservation first with equity-light tilts.", + "income": "Yield-focused mix emphasizing distributions and defensives.", + } + return descriptions.get(strategy, strategy) + + def _build_horizon_allocations( + self, + horizon_key: str, + allocations: Dict[str, AllocationSuggestion], + budget: float, + ) -> Dict[str, Any]: + total_weight = sum(item.weight for item in allocations.values()) + results = [] + cumulative = 0.0 + for asset_class, suggestion in allocations.items(): + weight = suggestion.weight / total_weight + amount = round(budget * weight, 2) + cumulative += amount + results.append( + { + "asset_class": asset_class, + "weight": round(weight, 3), + "amount": amount, + "rationale": suggestion.rationale, + "sample_assets": suggestion.sample_assets, + } + ) + drift = round(budget - cumulative, 2) + if abs(drift) >= 0.01 and results: + results[0]["amount"] = round(results[0]["amount"] + drift, 2) + + return { + "label": HORIZON_LABELS.get(horizon_key, horizon_key), + "allocations": results, + "risk_focus": self._risk_focus(horizon_key), + } + + def _risk_focus(self, horizon: str) -> str: + focus_map = { + "1w": "Liquidity & catalyst trading", + "1m": "Trend capture with guardrails", + "1y": "Compounding and thematic positioning", + } + return focus_map.get(horizon, "Balanced risk") + + +__all__ = ["MultiAssetAllocationAgent"] diff --git a/src/tradegraph_financial_advisor/main.py b/src/tradegraph_financial_advisor/main.py index bac70dc..24bd26d 100644 --- a/src/tradegraph_financial_advisor/main.py +++ b/src/tradegraph_financial_advisor/main.py @@ -9,12 +9,13 @@ from .agents.recommendation_engine import TradingRecommendationEngine from .agents.report_analysis_agent import ReportAnalysisAgent from .agents.channel_report_agent import ChannelReportAgent +from .agents.multi_asset_allocation_agent import MultiAssetAllocationAgent from .config.settings import settings, refresh_openai_api_key from .utils.helpers import save_analysis_results from .visualization import charts from .services.channel_stream_service import FinancialNewsChannelService from .services.price_trend_service import PriceTrendService -from .reporting import ChannelPDFReportWriter +from .reporting import ChannelPDFReportWriter, MultiAssetPDFReportWriter class FinancialAdvisor: @@ -29,6 +30,8 @@ def __init__(self, llm_model_name: str = "gpt-5-nano"): self.channel_service = FinancialNewsChannelService() self.trend_service = PriceTrendService() self.pdf_report_writer = ChannelPDFReportWriter() + self.multi_asset_agent = MultiAssetAllocationAgent() + self.multi_asset_pdf_writer = MultiAssetPDFReportWriter() async def analyze_portfolio( self, @@ -113,8 +116,10 @@ async def analyze_portfolio( "portfolio_recommendation": ( portfolio_recommendation if portfolio_recommendation else None ), + "recommendations": workflow_results.get("recommendations", []), "sentiment_analysis": sentiment_analysis, "detailed_reports": report_analyses, + "channel_streams": workflow_results.get("channel_streams", {}), "analysis_metadata": { "workflow_version": "1.0.0", "agents_used": [ @@ -183,6 +188,24 @@ async def quick_analysis( logger.error(f"Quick analysis failed: {str(e)}") raise + async def plan_multi_asset_allocation( + self, *, budget: float, strategies: Optional[List[str]] = None + ) -> Dict[str, Any]: + if budget <= 0: + raise ValueError("Budget must be positive for allocation planning.") + payload = { + "budget": budget, + "strategies": strategies, + } + return await self.multi_asset_agent.execute(payload) + + def build_multi_asset_pdf( + self, plan: Dict[str, Any], output_path: Optional[str] = None + ) -> str: + return self.multi_asset_pdf_writer.build_report( + plan=plan, output_path=output_path + ) + async def generate_channel_pdf_report( self, symbols: List[str], @@ -221,12 +244,24 @@ async def generate_channel_pdf_report( } ) + recommendations = reference_results.get("recommendations", []) + portfolio_rec = reference_results.get("portfolio_recommendation") + allocation_chart_path = None + if recommendations: + allocation_chart_path = charts.create_portfolio_allocation_chart( + recommendations=recommendations, + output_path="results/portfolio_allocation.png", + ) + pdf_path = self.pdf_report_writer.build_report( summary_payload=summary_payload, channel_payloads=channel_streams, price_trends=price_trends, - recommendations=reference_results.get("recommendations", []), + recommendations=recommendations, symbols=symbols, + portfolio_recommendation=portfolio_rec, + analysis_summary=reference_results.get("analysis_summary", {}), + allocation_chart_path=allocation_chart_path, output_path=output_path, ) @@ -341,6 +376,46 @@ def print_recommendations(self, results: Dict[str, Any]) -> None: print("\n" + "=" * 80) + def print_multi_asset_plan(self, plan: Dict[str, Any]) -> None: + budget = plan.get("budget", 0) + print("\n" + "=" * 80) + print("TRADEGRAPH MULTI-ASSET ALLOCATION PLAN") + print("=" * 80) + print(f"Budget: ${budget:,.2f}") + + strategies = plan.get("strategies", []) + for strategy in strategies: + print( + f"\n📌 Strategy: {strategy.get('strategy', '').title()} - {strategy.get('description', '')}" + ) + horizons = strategy.get("horizons", {}) + for horizon_key, payload in horizons.items(): + label = payload.get("label", horizon_key) + print(f" ➤ {label}: {payload.get('risk_focus', 'N/A')}") + for allocation in payload.get("allocations", []): + percent = allocation.get("weight", 0) * 100 + amount = allocation.get("amount", 0) + rationale = allocation.get("rationale", "") + sample_assets = ", ".join( + f"{asset['symbol']} ({asset['thesis']})" + for asset in allocation.get("sample_assets", []) + ) + print( + f" - {allocation.get('asset_class').upper()}: {percent:.1f}% " + f"(${amount:,.2f})" + ) + if rationale: + print(f" Rationale: {rationale}") + if sample_assets: + print(f" Sample: {sample_assets}") + + notes = plan.get("notes") or [] + if notes: + print("\n🗒 Advisor Notes:") + for note in notes: + print(f" - {note}") + print("\n" + "=" * 80) + async def main(): """ @@ -395,6 +470,21 @@ async def main(): type=str, help="Optional output path for the PDF report", ) + parser.add_argument( + "--multi-asset-budget", + type=float, + help="USD budget for a quick stocks/ETFs/crypto allocation plan", + ) + parser.add_argument( + "--multi-asset-strategies", + type=str, + help="Comma-separated strategies (growth,balanced,defensive,income)", + ) + parser.add_argument( + "--multi-asset-pdf-path", + type=str, + help="Optional output path for the multi-asset PDF report", + ) args = parser.parse_args() @@ -412,6 +502,34 @@ async def main(): try: advisor = FinancialAdvisor() + if args.multi_asset_budget: + strategies = None + if args.multi_asset_strategies: + strategies = [ + item.strip() + for item in args.multi_asset_strategies.split(",") + if item.strip() + ] + plan = await advisor.plan_multi_asset_allocation( + budget=args.multi_asset_budget, + strategies=strategies, + ) + try: + pdf_path = advisor.build_multi_asset_pdf( + plan, output_path=args.multi_asset_pdf_path + ) + logger.info(f"Multi-asset PDF saved to: {pdf_path}") + plan["pdf_path"] = pdf_path + except Exception as pdf_exc: + logger.warning(f"Failed to create multi-asset PDF: {pdf_exc}") + if args.output_format == "json": + import json + + print(json.dumps(plan, indent=2, default=str)) + else: + advisor.print_multi_asset_plan(plan) + return + if args.alerts_only: # Generate alerts only alerts = await advisor.get_stock_alerts(args.symbols) @@ -478,7 +596,7 @@ async def main(): chart_path = charts.create_portfolio_allocation_chart( recommendations=recommendations, - output_path="results/portfolio_allocation.html", + output_path="results/portfolio_allocation.png", ) logger.info(f"Portfolio allocation chart saved to: {chart_path}") diff --git a/src/tradegraph_financial_advisor/reporting/__init__.py b/src/tradegraph_financial_advisor/reporting/__init__.py index ca30cf1..7c9c915 100644 --- a/src/tradegraph_financial_advisor/reporting/__init__.py +++ b/src/tradegraph_financial_advisor/reporting/__init__.py @@ -1,5 +1,6 @@ """Reporting utilities for TradeGraph.""" from .pdf_reporter import ChannelPDFReportWriter +from .multi_asset_reporter import MultiAssetPDFReportWriter -__all__ = ["ChannelPDFReportWriter"] +__all__ = ["ChannelPDFReportWriter", "MultiAssetPDFReportWriter"] diff --git a/src/tradegraph_financial_advisor/reporting/multi_asset_reporter.py b/src/tradegraph_financial_advisor/reporting/multi_asset_reporter.py new file mode 100644 index 0000000..cde6dfe --- /dev/null +++ b/src/tradegraph_financial_advisor/reporting/multi_asset_reporter.py @@ -0,0 +1,162 @@ +"""PDF builder for multi-asset allocation plans.""" + +from __future__ import annotations + +import os +from datetime import datetime +from typing import Any, Dict, List, Optional +import textwrap + +from reportlab.lib.pagesizes import LETTER +from reportlab.lib.units import inch +from reportlab.pdfgen import canvas + + +class MultiAssetPDFReportWriter: + """Renders allocation plans across strategies/horizons into a PDF.""" + + def __init__(self) -> None: + self.page_width, self.page_height = LETTER + self.margin = 0.75 * inch + self.line_height = 14 + + def build_report( + self, + *, + plan: Dict[str, Any], + output_path: Optional[str] = None, + ) -> str: + os.makedirs("results", exist_ok=True) + if not output_path: + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + output_path = os.path.join( + "results", f"tradegraph_multi_asset_{timestamp}.pdf" + ) + + doc = canvas.Canvas(output_path, pagesize=LETTER) + cursor_y = self.page_height - self.margin + + cursor_y = self._draw_title(doc, cursor_y, "Multi-Asset Allocation Blueprint") + cursor_y = self._draw_subtitle( + doc, + cursor_y, + f"Budget: ${plan.get('budget', 0):,.2f} | Generated {datetime.now():%Y-%m-%d %H:%M UTC}", + ) + + notes = plan.get("notes") or [] + if notes: + cursor_y = self._draw_bullet_section( + doc, cursor_y, "Advisor Notes", notes + ) + + for strategy in plan.get("strategies", []): + cursor_y = self._draw_strategy_section(doc, cursor_y, strategy) + + doc.save() + return output_path + + def _draw_title(self, doc: canvas.Canvas, cursor_y: float, text: str) -> float: + doc.setFont("Helvetica-Bold", 20) + doc.drawString(self.margin, cursor_y, text) + return cursor_y - 24 + + def _draw_subtitle(self, doc: canvas.Canvas, cursor_y: float, text: str) -> float: + doc.setFont("Helvetica", 11) + doc.drawString(self.margin, cursor_y, text) + return cursor_y - 18 + + def _draw_strategy_section( + self, + doc: canvas.Canvas, + cursor_y: float, + strategy: Dict[str, Any], + ) -> float: + cursor_y = self._ensure_space(doc, cursor_y, min_height=140) + title = ( + f"Strategy: {str(strategy.get('strategy', '')).title()} - " + f"{strategy.get('description', '')}" + ) + doc.setFont("Helvetica-Bold", 14) + doc.drawString(self.margin, cursor_y, title) + cursor_y -= 18 + + horizons = strategy.get("horizons", {}) + for horizon_key, payload in horizons.items(): + cursor_y = self._draw_horizon(doc, cursor_y, horizon_key, payload) + return cursor_y + + def _draw_horizon( + self, + doc: canvas.Canvas, + cursor_y: float, + horizon_key: str, + payload: Dict[str, Any], + ) -> float: + cursor_y = self._ensure_space(doc, cursor_y, min_height=80) + label = payload.get("label", horizon_key) + doc.setFont("Helvetica-Bold", 12) + doc.drawString( + self.margin, + cursor_y, + f"{label}: {payload.get('risk_focus', 'Risk focus n/a')}", + ) + cursor_y -= 14 + doc.setFont("Helvetica", 10) + for allocation in payload.get("allocations", []): + amount = allocation.get("amount", 0) + weight = allocation.get("weight", 0) * 100 + doc.drawString( + self.margin + 10, + cursor_y, + f"- {allocation.get('asset_class', '').upper()}: {weight:.1f}% (${amount:,.2f})", + ) + cursor_y -= self.line_height + rationale = allocation.get("rationale") + if rationale: + for line in self._wrap_text(f"Rationale: {rationale}", 92): + doc.drawString(self.margin + 20, cursor_y, line) + cursor_y -= self.line_height + samples = allocation.get("sample_assets", []) + if samples: + sample_line = ", ".join( + f"{item.get('symbol')} ({item.get('thesis')})" for item in samples + ) + for line in self._wrap_text(f"Sample: {sample_line}", 92): + doc.drawString(self.margin + 20, cursor_y, line) + cursor_y -= self.line_height + cursor_y -= 4 + return cursor_y + + def _draw_bullet_section( + self, + doc: canvas.Canvas, + cursor_y: float, + title: str, + bullets: List[str], + ) -> float: + cursor_y = self._ensure_space(doc, cursor_y, min_height=70) + doc.setFont("Helvetica-Bold", 14) + doc.drawString(self.margin, cursor_y, title) + cursor_y -= 18 + doc.setFont("Helvetica", 11) + for bullet in bullets: + for line in self._wrap_text(bullet, 94): + doc.drawString(self.margin + 10, cursor_y, f"• {line}") + cursor_y -= self.line_height + return cursor_y - 6 + + def _ensure_space( + self, doc: canvas.Canvas, cursor_y: float, *, min_height: float + ) -> float: + if cursor_y - min_height <= self.margin: + doc.showPage() + cursor_y = self.page_height - self.margin + return cursor_y + + def _wrap_text(self, text: str, width: int) -> List[str]: + if not text: + return [""] + return textwrap.wrap(text, width=width) or [text] + + +__all__ = ["MultiAssetPDFReportWriter"] diff --git a/tests/unit/test_multi_asset_agent.py b/tests/unit/test_multi_asset_agent.py new file mode 100644 index 0000000..009de97 --- /dev/null +++ b/tests/unit/test_multi_asset_agent.py @@ -0,0 +1,57 @@ +import pytest + +from tradegraph_financial_advisor.agents.multi_asset_allocation_agent import ( + MultiAssetAllocationAgent, +) +from tradegraph_financial_advisor.reporting import MultiAssetPDFReportWriter + + +@pytest.mark.asyncio +async def test_multi_asset_agent_returns_balanced_plan(): + agent = MultiAssetAllocationAgent() + result = await agent.execute({"budget": 10000, "strategies": ["growth", "unknown"]}) + + assert result["budget"] == 10000.0 + assert result["strategies"], "Strategies should not be empty" + plan = result["strategies"][0] + assert plan["strategy"] == "growth" + horizon = plan["horizons"]["1w"] + weights = sum(item["weight"] for item in horizon["allocations"]) + amounts = sum(item["amount"] for item in horizon["allocations"]) + assert pytest.approx(weights, rel=1e-3) == 1.0 + assert pytest.approx(amounts, rel=1e-3) == 10000.0 + + +def test_multi_asset_pdf_writer(tmp_path): + writer = MultiAssetPDFReportWriter() + plan = { + "budget": 5000, + "strategies": [ + { + "strategy": "balanced", + "description": "Balanced mix", + "horizons": { + "1w": { + "label": "1-Week", + "risk_focus": "Liquidity", + "allocations": [ + { + "asset_class": "stocks", + "weight": 0.5, + "amount": 2500, + "rationale": "Test rationale", + "sample_assets": [ + {"symbol": "AAPL", "thesis": "Quality"} + ], + } + ], + } + }, + } + ], + "notes": ["Note"], + } + output_file = tmp_path / "multi_asset.pdf" + pdf_path = writer.build_report(plan=plan, output_path=str(output_file)) + assert output_file.exists() + assert pdf_path == str(output_file) From 0db085f06e60fb6522a599c8eea611f415476380 Mon Sep 17 00:00:00 2001 From: Mehran Moazeni Date: Tue, 9 Dec 2025 22:28:19 +0100 Subject: [PATCH 02/11] Multi-asset allocation + PDF --- pyproject.toml | 2 + requirements.txt | 2 + .../agents/channel_report_agent.py | 160 ++++++++++++++-- .../config/settings.py | 1 + .../reporting/pdf_reporter.py | 171 ++++++++++++++++- .../repositories/__init__.py | 3 + .../repositories/news_repository.py | 172 ++++++++++++++++++ .../services/channel_stream_service.py | 17 +- .../services/local_scraping_service.py | 44 ++++- .../visualization/charts.py | 20 +- tests/unit/test_channels.py | 20 ++ tests/unit/test_news_repository.py | 45 +++++ tradegraph.duckdb | Bin 1585152 -> 1585152 bytes uv.lock | 157 ++++++++++++++++ 14 files changed, 787 insertions(+), 27 deletions(-) create mode 100644 src/tradegraph_financial_advisor/repositories/__init__.py create mode 100644 src/tradegraph_financial_advisor/repositories/news_repository.py create mode 100644 tests/unit/test_news_repository.py diff --git a/pyproject.toml b/pyproject.toml index b647f71..073973d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,6 +41,8 @@ dependencies = [ "alpha-vantage>=2.3.0", "python-dateutil>=2.8.0", "plotly>=5.15.0", + "kaleido>=0.2.1", + "duckdb>=0.10.0", "fastapi>=0.118.0", "reportlab>=4.0.0", "uvicorn>=0.30.0", diff --git a/requirements.txt b/requirements.txt index 4a398b8..f1e231c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -19,5 +19,7 @@ pytest-asyncio>=0.21.0 pytest-cov>=4.1.0 python-dateutil>=2.8.0 plotly>=5.15.0 +kaleido>=0.2.1 +duckdb>=0.10.0 reportlab>=4.0.0 uvicorn>=0.30.0 diff --git a/src/tradegraph_financial_advisor/agents/channel_report_agent.py b/src/tradegraph_financial_advisor/agents/channel_report_agent.py index 03ac33c..4bff574 100644 --- a/src/tradegraph_financial_advisor/agents/channel_report_agent.py +++ b/src/tradegraph_financial_advisor/agents/channel_report_agent.py @@ -39,7 +39,9 @@ def __init__( ) async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: - channel_payloads = input_data.get("channel_payloads", {}) + channel_payloads = self._filter_channel_payloads( + input_data.get("channel_payloads", {}) + ) price_trends = input_data.get("price_trends", {}) recommendations = input_data.get("recommendations", []) @@ -54,14 +56,18 @@ async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: prompt_payload = { "channel_payloads": channel_payloads, "price_trends": price_trends, - "recommendations": recommendations, + "recommendation_snapshot": self._summarize_recommendations( + recommendations + ), } prompt = ( - "You are TradeGraph's senior analyst. Combine the multichannel news feeds, " - "risk context, and trend data below into a concise JSON summary with " - "keys: news_takeaways (list of strings), risk_assessment (string), " - "buy_or_sell_view (string), trend_commentary (string), key_stats (object), " - "and summary_text (string)." + "You are TradeGraph's senior portfolio strategist. Blend the curated news " + "feeds, short-term price action, and active recommendations into guidance " + "for an informed investor. Respond in JSON with keys: summary_text (2 " + "advisor-style paragraphs), advisor_memo (actionable paragraph), " + "news_takeaways (list of 3 strings), guidance_points (list of next " + "actions), risk_assessment, buy_or_sell_view, trend_commentary, " + "price_action_notes (list), and key_stats (object describing counts)." ) response = await self.llm.ainvoke( [HumanMessage(content=f"{prompt}\nINPUT:\n{json.dumps(prompt_payload)[:6000]}")] @@ -89,22 +95,41 @@ def _build_fallback_summary( f"{payload.get('title', channel_id)} highlights: {titles}" ) + key_stats = self._build_key_stats(channel_payloads, recommendations) + risk_counts: Dict[str, int] = {} for rec in recommendations: risk = rec.get("risk_level", "unknown") risk_counts[risk] = risk_counts.get(risk, 0) + 1 buy_view = "hold" - buy_votes = sum(1 for rec in recommendations if "buy" in str(rec.get("recommendation", "")).lower()) - sell_votes = sum(1 for rec in recommendations if "sell" in str(rec.get("recommendation", "")).lower()) + buy_votes = sum( + 1 + for rec in recommendations + if "buy" in str(rec.get("recommendation", "")).lower() + ) + sell_votes = sum( + 1 + for rec in recommendations + if "sell" in str(rec.get("recommendation", "")).lower() + ) if buy_votes > sell_votes: buy_view = "buy" elif sell_votes > buy_votes: buy_view = "reduce" trend_commentary = self._summarize_trends(price_trends) - summary_text = generate_summary(" ".join(news_takeaways)) or ( - "Latest headlines aggregated across equity, crypto, and free news agencies." + price_action_notes = self._build_price_notes(price_trends) + guidance_points = self._build_guidance_points(recommendations, price_action_notes) + + narrative_seed = " ".join(news_takeaways[:3]) or "Mixed market color." + summary_text = ( + generate_summary(narrative_seed) + or "Fresh headlines suggest a balanced tape across equities and crypto." + ) + advisor_memo = ( + f"My read: {buy_view.upper()} bias while monitoring {risk_counts or {'unknown': 0}}. " + f"Trend check: {trend_commentary}." ) return { @@ -112,11 +137,116 @@ def _build_fallback_summary( "risk_assessment": f"Risk mix: {risk_counts or {'unknown': 0}}", "buy_or_sell_view": buy_view, "trend_commentary": trend_commentary, - "key_stats": { - "recommendation_count": len(recommendations), - "channels": list(channel_payloads.keys()), - }, + "key_stats": key_stats, "summary_text": summary_text, + "advisor_memo": advisor_memo, + "price_action_notes": price_action_notes, + "guidance_points": guidance_points, + } + + def _filter_channel_payloads( + self, channel_payloads: Dict[str, Any] + ) -> Dict[str, Any]: + return { + channel_id: payload + for channel_id, payload in channel_payloads.items() + if channel_id != "open_source_agencies" + } + + def _build_key_stats( + self, + channel_payloads: Dict[str, Any], + recommendations: List[Dict[str, Any]], + ) -> Dict[str, Any]: + total_items = sum(len(payload.get("items", [])) for payload in channel_payloads.values()) + covered_symbols = { + symbol + for payload in channel_payloads.values() + for item in payload.get("items", []) + for symbol in item.get("matched_symbols", []) + if symbol + } + return { + "channel_count": len(channel_payloads), + "headline_count": total_items, + "recommendation_count": len(recommendations), + "covered_symbols": sorted(covered_symbols), + } + + def _build_price_notes(self, price_trends: Dict[str, Any]) -> List[str]: + notes: List[str] = [] + for symbol, payload in price_trends.items(): + trends = payload.get("trends", {}) + day = trends.get("last_day", {}).get("percent_change") + week = trends.get("last_week", {}).get("percent_change") + hour = trends.get("last_hour", {}).get("percent_change") + pieces = [] + if week is not None: + pieces.append(f"{week:+.1f}% weekly") + if day is not None: + pieces.append(f"{day:+.1f}% daily") + if hour is not None: + pieces.append(f"{hour:+.1f}% hourly") + if pieces: + notes.append(f"{symbol}: {' / '.join(pieces)} post-close move") + return notes + + def _build_guidance_points( + self, + recommendations: List[Dict[str, Any]], + price_notes: List[str], + ) -> List[str]: + guidance: List[str] = [] + for rec in recommendations[:3]: + symbol = rec.get("symbol", "") + rec_text = rec.get("recommendation", "hold").replace("_", " ") + allocation = rec.get("recommended_allocation") + allocation_text = ( + f"targeting {allocation:.1%} weight" + if isinstance(allocation, (int, float)) + else "" + ) + note = rec.get("analyst_notes") or ", ".join(rec.get("key_factors", [])[:2]) + clause = f"{symbol}: {rec_text.title()} {allocation_text}".strip() + if note: + clause = f"{clause} — {note}" + guidance.append(clause) + + if price_notes: + guidance.append(f"Monitor price tape: {price_notes[0]}") + return guidance + + def _summarize_recommendations( + self, recommendations: List[Dict[str, Any]] + ) -> Dict[str, Any]: + if not recommendations: + return {"total": 0} + + counts: Dict[str, int] = {} + top_symbols: List[str] = [] + highest_conf = sorted( + recommendations, + key=lambda rec: rec.get("confidence_score", 0), + reverse=True, + )[:3] + for rec in recommendations: + name = str(rec.get("recommendation", "unknown")).lower() + counts[name] = counts.get(name, 0) + 1 + if rec.get("symbol"): + top_symbols.append(rec["symbol"]) + + return { + "total": len(recommendations), + "counts": counts, + "top_conviction": [ + { + "symbol": rec.get("symbol"), + "confidence_score": rec.get("confidence_score"), + "recommendation": rec.get("recommendation"), + } + for rec in highest_conf + ], + "symbols": top_symbols, } def _summarize_trends(self, price_trends: Dict[str, Any]) -> str: diff --git a/src/tradegraph_financial_advisor/config/settings.py b/src/tradegraph_financial_advisor/config/settings.py index b586145..1f3c0fa 100644 --- a/src/tradegraph_financial_advisor/config/settings.py +++ b/src/tradegraph_financial_advisor/config/settings.py @@ -30,6 +30,7 @@ class Settings(BaseSettings): ) analysis_depth: str = Field("detailed", env="ANALYSIS_DEPTH") default_portfolio_size: float = Field(100000.0, env="DEFAULT_PORTFOLIO_SIZE") + news_db_path: str = Field("tradegraph.duckdb", env="NEWS_DB_PATH") model_config = {"env_file": ".env", "case_sensitive": False, "extra": "ignore"} diff --git a/src/tradegraph_financial_advisor/reporting/pdf_reporter.py b/src/tradegraph_financial_advisor/reporting/pdf_reporter.py index 1156c5e..a57cd64 100644 --- a/src/tradegraph_financial_advisor/reporting/pdf_reporter.py +++ b/src/tradegraph_financial_advisor/reporting/pdf_reporter.py @@ -9,6 +9,7 @@ from reportlab.lib.pagesizes import LETTER from reportlab.lib.units import inch +from reportlab.lib.utils import ImageReader from reportlab.pdfgen import canvas @@ -28,6 +29,9 @@ def build_report( price_trends: Dict[str, Any], recommendations: List[Dict[str, Any]], symbols: List[str], + portfolio_recommendation: Optional[Dict[str, Any]] = None, + analysis_summary: Optional[Dict[str, Any]] = None, + allocation_chart_path: Optional[str] = None, output_path: Optional[str] = None, ) -> str: os.makedirs("results", exist_ok=True) @@ -47,12 +51,42 @@ def build_report( f"Symbols: {', '.join(symbols)} | Generated {datetime.now():%Y-%m-%d %H:%M UTC}", ) + cursor_y = self._draw_portfolio_overview( + doc, + cursor_y, + analysis_summary or {}, + portfolio_recommendation or {}, + allocation_chart_path, + ) + cursor_y = self._draw_section( doc, cursor_y, "Executive Summary", summary_payload.get("summary_text", "") ) - news_text = "\n".join(summary_payload.get("news_takeaways", [])) - cursor_y = self._draw_section(doc, cursor_y, "News Highlights", news_text) + cursor_y = self._draw_section( + doc, cursor_y, "Desk Memo", summary_payload.get("advisor_memo", "") + ) + + cursor_y = self._draw_bullet_section( + doc, + cursor_y, + "News Highlights", + summary_payload.get("news_takeaways", []), + ) + + cursor_y = self._draw_bullet_section( + doc, + cursor_y, + "Price & Trend Signals", + summary_payload.get("price_action_notes", []), + ) + + cursor_y = self._draw_bullet_section( + doc, + cursor_y, + "Actionable Guidance", + summary_payload.get("guidance_points", []), + ) cursor_y = self._draw_section( doc, @@ -63,6 +97,8 @@ def build_report( f"Trend Notes: {summary_payload.get('trend_commentary', 'n/a')}", ) + cursor_y = self._draw_key_stats(doc, cursor_y, summary_payload.get("key_stats", {})) + cursor_y = self._draw_channel_breakdown(doc, cursor_y, channel_payloads) cursor_y = self._draw_recommendations(doc, cursor_y, recommendations) cursor_y = self._draw_trends(doc, cursor_y, price_trends) @@ -94,6 +130,52 @@ def _draw_section( cursor_y -= self.line_height return cursor_y - 6 + def _draw_bullet_section( + self, + doc: canvas.Canvas, + cursor_y: float, + title: str, + items: List[str], + ) -> float: + if not items: + return cursor_y + cursor_y = self._ensure_space(doc, cursor_y, min_height=80) + doc.setFont("Helvetica-Bold", 14) + doc.drawString(self.margin, cursor_y, title) + cursor_y -= 18 + doc.setFont("Helvetica", 11) + for item in items: + for line in self._wrap_text(item, 92): + doc.drawString(self.margin + 10, cursor_y, f"• {line}") + cursor_y -= self.line_height + return cursor_y - 6 + + def _draw_key_stats( + self, + doc: canvas.Canvas, + cursor_y: float, + stats: Dict[str, Any], + ) -> float: + if not stats: + return cursor_y + cursor_y = self._ensure_space(doc, cursor_y, min_height=70) + doc.setFont("Helvetica-Bold", 14) + doc.drawString(self.margin, cursor_y, "Monitoring Stats") + cursor_y -= 18 + doc.setFont("Helvetica", 11) + lines = [ + f"Channels monitored: {stats.get('channel_count', 0)}", + f"Headlines ingested: {stats.get('headline_count', 0)}", + f"Recommendations referenced: {stats.get('recommendation_count', 0)}", + ] + covered = stats.get("covered_symbols") + if covered: + lines.append(f"Symbols highlighted: {', '.join(covered)}") + for line in lines: + doc.drawString(self.margin, cursor_y, line) + cursor_y -= self.line_height + return cursor_y - 6 + def _draw_channel_breakdown( self, doc: canvas.Canvas, cursor_y: float, channels: Dict[str, Any] ) -> float: @@ -184,6 +266,82 @@ def _draw_trends( cursor_y -= self.line_height return cursor_y + def _draw_portfolio_overview( + self, + doc: canvas.Canvas, + cursor_y: float, + analysis_summary: Dict[str, Any], + portfolio_recommendation: Dict[str, Any], + allocation_chart_path: Optional[str], + ) -> float: + cursor_y = self._ensure_space(doc, cursor_y, min_height=180) + section_top = cursor_y + doc.setFont("Helvetica-Bold", 14) + doc.drawString(self.margin, cursor_y, "Portfolio Overview") + cursor_y -= 18 + doc.setFont("Helvetica", 11) + + portfolio_size = analysis_summary.get("portfolio_size") + risk_tolerance = analysis_summary.get("risk_tolerance", "-") + time_horizon = analysis_summary.get("time_horizon", "-") + symbols_line = ", ".join(analysis_summary.get("symbols_analyzed", [])) + text_lines = [ + f"Portfolio Size: {self._format_currency(portfolio_size)}", + f"Risk Tolerance: {risk_tolerance.title() if isinstance(risk_tolerance, str) else risk_tolerance}", + f"Time Horizon: {time_horizon.replace('_', ' ').title() if isinstance(time_horizon, str) else time_horizon}", + ] + if symbols_line: + text_lines.append(f"Focus Symbols: {symbols_line}") + + total_conf = portfolio_recommendation.get("total_confidence") + diversification = portfolio_recommendation.get("diversification_score") + expected_return = portfolio_recommendation.get("expected_return") + expected_vol = portfolio_recommendation.get("expected_volatility") + overall_risk = portfolio_recommendation.get("overall_risk_level") + if isinstance(total_conf, (int, float)): + text_lines.append(f"Portfolio Confidence: {total_conf:.0%}") + if isinstance(diversification, (int, float)): + text_lines.append(f"Diversification Score: {diversification:.0%}") + if isinstance(expected_return, (int, float)): + text_lines.append(f"Expected Return: {expected_return:.1%}") + elif expected_return: + text_lines.append(f"Expected Return: {expected_return}") + if isinstance(expected_vol, (int, float)): + text_lines.append(f"Expected Volatility: {expected_vol:.1%}") + elif expected_vol: + text_lines.append(f"Expected Volatility: {expected_vol}") + if overall_risk: + text_lines.append(f"Overall Risk: {str(overall_risk).title()}") + + for line in text_lines: + doc.drawString(self.margin, cursor_y, line) + cursor_y -= self.line_height + + cursor_after_text = cursor_y - 6 + + chart_bottom = cursor_after_text + if allocation_chart_path and os.path.exists(allocation_chart_path): + try: + image = ImageReader(allocation_chart_path) + chart_width = 2.8 * inch + chart_height = 2.8 * inch + chart_x = self.page_width - self.margin - chart_width + chart_y = section_top - chart_height + doc.drawImage( + image, + chart_x, + chart_y, + width=chart_width, + height=chart_height, + preserveAspectRatio=True, + mask="auto", + ) + chart_bottom = min(chart_y - 10, cursor_after_text) + except Exception: + chart_bottom = cursor_after_text + + return min(cursor_after_text, chart_bottom) + def _ensure_space( self, doc: canvas.Canvas, cursor_y: float, *, min_height: float ) -> float: @@ -206,5 +364,14 @@ def _format_pct(trend: Optional[Dict[str, Any]]) -> str: return " - " return f"{percent:+.1f}%" + @staticmethod + def _format_currency(value: Optional[float]) -> str: + if value is None: + return "n/a" + try: + return f"${float(value):,.0f}" + except (TypeError, ValueError): + return str(value) + __all__ = ["ChannelPDFReportWriter"] diff --git a/src/tradegraph_financial_advisor/repositories/__init__.py b/src/tradegraph_financial_advisor/repositories/__init__.py new file mode 100644 index 0000000..f20f5a3 --- /dev/null +++ b/src/tradegraph_financial_advisor/repositories/__init__.py @@ -0,0 +1,3 @@ +from .news_repository import NewsRepository + +__all__ = ["NewsRepository"] diff --git a/src/tradegraph_financial_advisor/repositories/news_repository.py b/src/tradegraph_financial_advisor/repositories/news_repository.py new file mode 100644 index 0000000..03d665f --- /dev/null +++ b/src/tradegraph_financial_advisor/repositories/news_repository.py @@ -0,0 +1,172 @@ +"""DuckDB-backed persistence for scraped news articles.""" + +from __future__ import annotations + +import threading +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional, Sequence, Union + +import duckdb +from dateutil import parser as date_parser +from loguru import logger +from pydantic import ValidationError + +from ..config.settings import settings +from ..models.financial_data import NewsArticle + + +class NewsRepository: + """Simple repository that stores news articles inside DuckDB.""" + + def __init__(self, db_path: Optional[Union[str, Path]] = None) -> None: + path = Path(db_path or settings.news_db_path).expanduser() + if not path.is_absolute(): + path = Path.cwd() / path + path.parent.mkdir(parents=True, exist_ok=True) + self.db_path = path + self._schema_lock = threading.Lock() + self._write_lock = threading.Lock() + self._ensure_schema() + + def _ensure_schema(self) -> None: + with self._schema_lock: + with duckdb.connect(str(self.db_path)) as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS news_articles ( + symbol TEXT, + title TEXT, + url TEXT, + summary TEXT, + content TEXT, + source TEXT, + published_at TIMESTAMP, + scraped_at TIMESTAMP, + symbols TEXT, + sentiment TEXT, + impact_score DOUBLE + ); + """ + ) + conn.execute( + """ + CREATE UNIQUE INDEX IF NOT EXISTS idx_news_articles_symbol_title_url + ON news_articles(symbol, title, url); + """ + ) + + def record_articles( + self, articles: Sequence[Union[NewsArticle, Dict[str, Any]]] + ) -> int: + rows: List[tuple] = [] + scraped_at = datetime.utcnow() + for article in articles: + model = self._coerce_article(article) + if not model: + continue + primary_symbol = model.symbols[0] if model.symbols else None + rows.append( + ( + primary_symbol, + model.title.strip(), + model.url, + (model.summary or "").strip() or None, + model.content, + model.source, + self._normalize_datetime(model.published_at), + scraped_at, + ",".join(model.symbols), + self._normalize_sentiment(model.sentiment), + model.impact_score, + ) + ) + + if not rows: + return 0 + + insert_sql = """ + INSERT INTO news_articles ( + symbol, + title, + url, + summary, + content, + source, + published_at, + scraped_at, + symbols, + sentiment, + impact_score + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(symbol, title, url) DO UPDATE SET + summary=excluded.summary, + content=excluded.content, + source=excluded.source, + published_at=excluded.published_at, + scraped_at=excluded.scraped_at, + symbols=excluded.symbols, + sentiment=excluded.sentiment, + impact_score=excluded.impact_score; + """ + + with self._write_lock: + try: + with duckdb.connect(str(self.db_path)) as conn: + conn.executemany(insert_sql, rows) + except Exception as exc: # pragma: no cover - disk/config issues + logger.warning(f"Failed to persist news articles: {exc}") + return 0 + + return len(rows) + + def fetch_recent_articles(self, limit: int = 50) -> List[Dict[str, Any]]: + query = ( + "SELECT symbol, title, url, summary, source, published_at, scraped_at, symbols, sentiment, impact_score " + "FROM news_articles ORDER BY COALESCE(published_at, scraped_at) DESC LIMIT ?" + ) + with duckdb.connect(str(self.db_path)) as conn: + rows = conn.execute(query, [limit]).fetchall() + columns = [desc[0] for desc in conn.description] + + return [dict(zip(columns, row)) for row in rows] + + def _coerce_article( + self, article: Union[NewsArticle, Dict[str, Any], None] + ) -> Optional[NewsArticle]: + if article is None: + return None + if isinstance(article, NewsArticle): + return article + if isinstance(article, dict): + try: + return NewsArticle(**article) + except ValidationError as exc: + logger.warning(f"Invalid news article payload skipped: {exc}") + return None + logger.warning("Unsupported news article type {}", type(article)) + return None + + @staticmethod + def _normalize_datetime(value: Any) -> Optional[datetime]: + if isinstance(value, datetime): + return value + if not value: + return None + if isinstance(value, str): + try: + return date_parser.parse(value) + except (ValueError, TypeError): + return None + return None + + @staticmethod + def _normalize_sentiment(value: Any) -> Optional[str]: + if value is None: + return None + if hasattr(value, "value"): + return str(value.value) + return str(value) + + +__all__ = ["NewsRepository"] diff --git a/src/tradegraph_financial_advisor/services/channel_stream_service.py b/src/tradegraph_financial_advisor/services/channel_stream_service.py index b828d7f..5f388f5 100644 --- a/src/tradegraph_financial_advisor/services/channel_stream_service.py +++ b/src/tradegraph_financial_advisor/services/channel_stream_service.py @@ -203,10 +203,12 @@ def __init__( max_items_per_source: int = 5, max_items_per_channel: int = 25, price_service: Optional[PriceTrendService] = None, + include_open_agencies: bool = False, ) -> None: self.max_items_per_source = max_items_per_source self.max_items_per_channel = max_items_per_channel self.price_service = price_service or PriceTrendService() + self.include_open_agencies = include_open_agencies self._session_lock = asyncio.Lock() self._session: Optional[aiohttp.ClientSession] = None @@ -222,19 +224,19 @@ async def _get_session(self) -> aiohttp.ClientSession: return self._session def describe_channels(self) -> List[Dict[str, Any]]: - return [definition.metadata() for definition in CHANNEL_REGISTRY.values()] + return [CHANNEL_REGISTRY[channel].metadata() for channel in self._iter_channels()] async def collect_all_channels( self, symbols: Optional[Sequence[str]] = None ) -> Dict[str, Any]: tasks = [ self.fetch_channel_payload(channel.value, symbols) - for channel in ChannelType + for channel in self._iter_channels() ] payloads = await asyncio.gather(*tasks, return_exceptions=True) result: Dict[str, Any] = {} - for channel, payload in zip(ChannelType, payloads): + for channel, payload in zip(self._iter_channels(), payloads): if isinstance(payload, Exception): logger.warning( f"Failed to collect channel {channel.value}: {payload}" @@ -243,6 +245,15 @@ async def collect_all_channels( result[channel.value] = payload return result + def _iter_channels(self): + for channel in ChannelType: + if ( + not self.include_open_agencies + and channel == ChannelType.OPEN_SOURCE_AGENCIES + ): + continue + yield channel + async def fetch_channel_payload( self, channel_id: str, symbols: Optional[Sequence[str]] = None ) -> Dict[str, Any]: diff --git a/src/tradegraph_financial_advisor/services/local_scraping_service.py b/src/tradegraph_financial_advisor/services/local_scraping_service.py index 2ba7974..6141740 100644 --- a/src/tradegraph_financial_advisor/services/local_scraping_service.py +++ b/src/tradegraph_financial_advisor/services/local_scraping_service.py @@ -3,7 +3,8 @@ from __future__ import annotations import asyncio -from typing import Any, Dict, List, Sequence +from typing import Any, Dict, List, Sequence, Optional +from urllib.parse import urlparse from loguru import logger from ddgs import DDGS @@ -11,11 +12,23 @@ from ..models.financial_data import NewsArticle from ..utils.helpers import generate_summary +from ..repositories import NewsRepository class LocalScrapingService: - def __init__(self): + OPEN_AGENCY_DOMAINS = { + "theguardian.com", + "guardian.co.uk", + "bbc.co.uk", + "bbci.co.uk", + "aljazeera.com", + "npr.org", + "financialexpress.com", + } + + def __init__(self, news_repository: Optional[NewsRepository] = None): self.crawler = AsyncWebCrawler() + self.news_repository = news_repository or self._build_repository() async def search_and_scrape_news( self, symbols: Sequence[str], max_articles_per_symbol: int = 5 @@ -26,6 +39,8 @@ async def search_and_scrape_news( if not symbols: return all_articles + new_articles: List[NewsArticle] = [] + with DDGS() as ddgs: for symbol in symbols: query = f"{symbol} stock news" @@ -47,6 +62,12 @@ async def search_and_scrape_news( url = result.get("url") or result.get("href") if not url: continue + netloc = urlparse(url).netloc.lower() + if any( + netloc.endswith(domain) + for domain in self.OPEN_AGENCY_DOMAINS + ): + continue scraped_data = await self.crawler.arun(url) if not scraped_data or not scraped_data.markdown: continue @@ -60,9 +81,12 @@ async def search_and_scrape_news( symbols=[symbol], ) all_articles.append(article) + new_articles.append(article) except Exception as scrape_exc: logger.warning(f"Failed to scrape article {result.get('url')}: {scrape_exc}") + if new_articles: + await self._persist_articles(new_articles) return all_articles async def search_and_scrape_financial_reports( @@ -112,6 +136,7 @@ async def start(self): async def stop(self): logger.info("LocalScrapingService stopped.") await self.crawler.close() + self.news_repository = None async def health_check(self) -> bool: # For now, we assume the service is healthy if it can be instantiated. @@ -136,3 +161,18 @@ def _runner() -> List[Dict[str, Any]]: return list(ddgs.text(query, **kwargs)) return await asyncio.to_thread(_runner) + + def _build_repository(self) -> Optional[NewsRepository]: + try: + return NewsRepository() + except Exception as exc: + logger.warning(f"News repository initialization failed: {exc}") + return None + + async def _persist_articles(self, articles: List[NewsArticle]) -> None: + if not self.news_repository or not articles: + return + try: + await asyncio.to_thread(self.news_repository.record_articles, articles) + except Exception as exc: + logger.warning(f"Failed to write scraped news to DuckDB: {exc}") diff --git a/src/tradegraph_financial_advisor/visualization/charts.py b/src/tradegraph_financial_advisor/visualization/charts.py index c75fd53..e22794a 100644 --- a/src/tradegraph_financial_advisor/visualization/charts.py +++ b/src/tradegraph_financial_advisor/visualization/charts.py @@ -1,8 +1,10 @@ +from pathlib import Path + import plotly.graph_objects as go def create_portfolio_allocation_chart( - recommendations, output_path="portfolio_allocation.html" + recommendations, output_path="results/portfolio_allocation.png" ): """ Creates a pie chart showing portfolio allocation @@ -43,8 +45,16 @@ def create_portfolio_allocation_chart( fig.update_layout(title="Portfolio Allocation Recommendation", showlegend=True) - # Save to HTML file - fig.write_html(output_path) - print(f"Chart saved to: {output_path}") + path = Path(output_path) + path.parent.mkdir(parents=True, exist_ok=True) + + try: + fig.write_image(str(path)) + except ValueError as exc: + raise RuntimeError( + "Plotly static image export requires the kaleido package." + ) from exc + + print(f"Chart saved to: {path}") - return output_path + return str(path) diff --git a/tests/unit/test_channels.py b/tests/unit/test_channels.py index 004f899..a18970f 100644 --- a/tests/unit/test_channels.py +++ b/tests/unit/test_channels.py @@ -43,6 +43,9 @@ async def test_channel_report_agent_fallback( ) assert summary["news_takeaways"] assert "summary_text" in summary + assert summary.get("advisor_memo") + assert summary.get("guidance_points") + assert summary.get("key_stats", {}).get("channel_count") == 1 def test_pdf_report_writer(tmp_path, sample_channel_streams, sample_price_trends, sample_recommendations): @@ -54,6 +57,10 @@ def test_pdf_report_writer(tmp_path, sample_channel_streams, sample_price_trends "risk_assessment": "medium", "buy_or_sell_view": "buy", "trend_commentary": "AAPL: +5% YoY", + "advisor_memo": "Maintain constructive stance with risk controls.", + "price_action_notes": ["AAPL: +2.5% weekly"], + "guidance_points": ["Add MSFT on earnings strength"], + "key_stats": {"channel_count": 1, "headline_count": 2, "recommendation_count": 2}, } pdf_path = writer.build_report( summary_payload=summary_payload, @@ -61,6 +68,19 @@ def test_pdf_report_writer(tmp_path, sample_channel_streams, sample_price_trends price_trends=sample_price_trends, recommendations=sample_recommendations, symbols=["AAPL"], + analysis_summary={ + "portfolio_size": 100000, + "risk_tolerance": "medium", + "time_horizon": "medium_term", + "symbols_analyzed": ["AAPL"], + }, + portfolio_recommendation={ + "total_confidence": 0.8, + "diversification_score": 0.7, + "expected_return": 0.12, + "expected_volatility": 0.2, + "overall_risk_level": "medium", + }, output_path=str(output_file), ) assert os.path.exists(pdf_path) diff --git a/tests/unit/test_news_repository.py b/tests/unit/test_news_repository.py new file mode 100644 index 0000000..a8c79c1 --- /dev/null +++ b/tests/unit/test_news_repository.py @@ -0,0 +1,45 @@ +from datetime import datetime, timezone + +import pytest + +from tradegraph_financial_advisor.models.financial_data import NewsArticle +from tradegraph_financial_advisor.repositories import NewsRepository + + +def _sample_article(**overrides): + base = { + "title": "Sample headline", + "url": "https://example.com/story", + "content": "Detailed article body", + "summary": "Summary", + "source": "ExampleWire", + "published_at": datetime(2024, 1, 1, tzinfo=timezone.utc), + "symbols": ["AAPL"], + } + base.update(overrides) + return NewsArticle(**base) + + +def test_news_repository_upsert(tmp_path): + repo = NewsRepository(db_path=tmp_path / "news.duckdb") + + first = _sample_article() + repo.record_articles([first]) + + # Update summary to ensure UPSERT semantics + updated = _sample_article(summary="Updated summary") + repo.record_articles([updated]) + + rows = repo.fetch_recent_articles(limit=10) + assert len(rows) == 1 + assert rows[0]["summary"] == "Updated summary" + assert rows[0]["symbol"] == "AAPL" + + +def test_news_repository_handles_invalid_articles(tmp_path): + repo = NewsRepository(db_path=tmp_path / "news.duckdb") + + inserted = repo.record_articles([None, {"title": "missing fields"}]) + + assert inserted == 0 + assert repo.fetch_recent_articles(limit=10) == [] diff --git a/tradegraph.duckdb b/tradegraph.duckdb index 8c2725495a6faf9e7a93392b11ea5aeae8d72773..e006d2ae0326a9251a4a98917a5bc046f012e486 100644 GIT binary patch delta 2647 zcmcImYiv|S6rS0;uXbB@`=n(l-Vk}1pbj=9NQ0KNOHvZr0Bu4HVYlqPrK^3>Z7rzr zE{VBBqiJ^!KlF#D#Xn%8p#-D-#r2%IyPLADN=-az zcW2I=^L^)>nK`qGxkzFz@}4YzvHkH&wTHg{-Ax`4Uv~D0DF^2K-aqP&K2u*aa@I`E z?xbdD>+ogj;-AkxT~XSnj>N;AgF~@sj~bsM3i0v7Y>nm`iVgLsD$Hvw>nI`JgjkkB zq~XDyB}z|xxVN`+@Wj%EXn)_33V($#?s)(3V00;b$-wZDo>;tF)s%3jaI09%UL21O zb`Ge^u6g1odXMz?#Fuat#UX-N?{Z<3#CiuhqeJ0%w0}?pvpUDdU;nv$Pl-yrmeoaw z$e$`?eM~tXexP7sn>8$|6^7z;5w}8Wuv82`Hsi+`%HO!Tjqha(u1b}ht~2*+{fKHo zm7FX?cnRPwD@vRS0S;#uarSn``eKAUPm;7)UPhYshjz7x`hJ1!;xc$Qk$0fH{#oN&kcI3X{D>X<2$=lr|J2S8(I~Y+Ws=B6b?8 zl0;_e*DDDS&0fou*32aLI*e1#(oGoiN%x8Y(Cwxxt+mnT96Xc!?0?Rdk6-LakA!K5 z1Bp)*iFnuBIo*NWvgu|O_E-KKL404)Dy-BtFHg<7`K@>B@+Z$s-6@y2)#NibG`-}c ztIe`|r=7-^%{20UCMg**qYR35-60sw)c!dqKzP%UQTnH!*6{I;ru1PaZFKSQv50Wv z*Sqb3vCcaC7;j}X0WdMj!;}Y5lm@IpWie>v_{z)`^ zaSRQ(KOO<&3~zjE4DmM7T0W7SF@z05ivb3N10)*5QN-rXvB2Sx`MG94-*MMx@)r&{ zjm9nXIO3br0~2l|!KjC>F^(}BHfKk?ULJ|;wvjW*jFG{3FnN&}VgfN_VRLeAn5Dra z0I%X}J0u!}iZ$4#7?$wCM6!Oi0;`jQ&8>)U3u=R$!VM_0RQZC5;$;gcjO;KG94eH9 zvbXYsu^BgA3*l_F;buqIzAaY@hcp!GhLbzb1vPk3ZU^v@!Apb!iW~5(CqW2hC!7nWP^=@| ziHO}E)7nA0;D$Lg6x85;0+PO%*}m)P@6az1y1`NT|cCp_|(OBd$=3UgI~V%%S#cn5YAeiWFrKIB3= z18#G`NNBXC$d!GO9LBgtt7xNn|EZ$VQ_yqZAA65`8TS7${!E7qA$ hXC*o-)!8bY`E?f1S((ntbylIXN}W~dtXgNQ{{n$f6y5*; delta 1765 zcmd^9&1+m$6u;-qM`n^tXObpO`k|AelDg^%LlGCn7`l?!7K$he&O;mc8H=G zs9d{i3I|~a*b$NVCE1)i3 z*{R7trhW*f_40s&YW+-AjZZez`V49uc96ZU@y?${icrRkOaE1_vni^0%H<#DyS7X5 zRF$o&v3zQA9c?AW#%wy$mf}3Tw7sIeR7Lx>mD#v6TIVJ z$igdNF`1O-$hayy>-r~ugbGl8%`oDdMQ7VaR4pURJUT#KuO}6*=2ZSbTE8_1JKNFv z^Bx@TS{g|}PTe^2k#}tLQlQJTGh*Cq%f6|hUC90`_;uLz#=}W1x$-SJ>>`1xgir)Jv>`=uo6LDS^-U%}R zJ7;8l?6A$8n|r~?txnY3JXH-Io&SjrrL-m#LebE5_J-49>bygSr}Z-7Xynb&h7Xba z>k}`wD{sebCP_-TUN8fozj^~k4X#KSGt!objmTe%&cn7 zdjpzVzdn)h>OVhD(!f5m%MkYY2Xrp)bt?` z@si`>whqVQP=}*JU$|)@JSe=&Xi{kiKdN!PJ@1UBW1$(-ON+2I9(2_L-_1MAS%x+2 z&n&{DtY2FM{Xf0asQ&yCwEW?3I|2xJ6=CVoqY8#D0leB@Rd& Jl( Date: Tue, 9 Dec 2025 22:47:35 +0100 Subject: [PATCH 03/11] Fix lint warnings after multi-asset changes --- .../agents/multi_asset_allocation_agent.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py b/src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py index af1bb46..51af66a 100644 --- a/src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py +++ b/src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py @@ -3,7 +3,7 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List from loguru import logger From c2a88cfbc1f2c7ff24fe34e29b93ad630ed17b9a Mon Sep 17 00:00:00 2001 From: Mehran Moazeni <77067119+Mehranmzn@users.noreplygithub.com> Date: Tue, 9 Dec 2025 23:07:51 +0100 Subject: [PATCH 04/11] Modernize settings and serialization for Pydantic v2 --- src/tradegraph_financial_advisor/agents/financial_agent.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/tradegraph_financial_advisor/agents/financial_agent.py b/src/tradegraph_financial_advisor/agents/financial_agent.py index 826904f..4762a70 100644 --- a/src/tradegraph_financial_advisor/agents/financial_agent.py +++ b/src/tradegraph_financial_advisor/agents/financial_agent.py @@ -62,7 +62,7 @@ async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: else: market_data = await self._get_equity_market_data(symbol) symbol_data["market_data"] = ( - market_data.dict() if market_data else None + market_data.model_dump() if market_data else None ) if include_financials: @@ -73,7 +73,7 @@ async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: symbol, market_data ) symbol_data["financials"] = ( - financials.dict() if financials else None + financials.model_dump() if financials else None ) if include_technical: @@ -82,7 +82,7 @@ async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: else: technical = await self._get_equity_technical_indicators(symbol) symbol_data["technical_indicators"] = ( - technical.dict() if technical else None + technical.model_dump() if technical else None ) results[symbol] = symbol_data From 06862d895c645074b0ec84be9fa775c53e7f7458 Mon Sep 17 00:00:00 2001 From: Mehran Moazeni Date: Wed, 10 Dec 2025 21:54:55 +0100 Subject: [PATCH 05/11] Remove Finnhub-based tests --- tests/conftest.py | 46 +------------------ tests/unit/test_agents.py | 96 --------------------------------------- 2 files changed, 2 insertions(+), 140 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 2ab5f08..5116302 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,7 @@ import asyncio import pytest import os -from unittest.mock import Mock, AsyncMock +from unittest.mock import AsyncMock from datetime import datetime, timedelta from tradegraph_financial_advisor.models.financial_data import NewsArticle @@ -9,6 +9,7 @@ # Test configuration os.environ["OPENAI_API_KEY"] = "test-openai-key" os.environ["ALPHA_VANTAGE_API_KEY"] = "test-alpha-vantage-key" +os.environ["FINNHUB_API_KEY"] = "test-finnhub-key" os.environ["LOG_LEVEL"] = "DEBUG" @@ -262,49 +263,6 @@ def sample_price_trends(): } -@pytest.fixture -def mock_finnhub_client(): - """Mock Finnhub client for testing.""" - client = AsyncMock() - - client.get_quote.return_value = {"c": 195.5, "o": 193.0, "v": 25000000} - - async def get_candles(symbol, resolution, start, end): - base = 150.0 - closes = [base + i for i in range(160)] - return { - "s": "ok", - "c": closes, - "h": [value + 2 for value in closes], - "l": [value - 2 for value in closes], - } - - client.get_candles.side_effect = get_candles - client.get_company_profile.return_value = { - "name": "Apple Inc.", - "marketCapitalization": 3_000_000_000_000, - } - client.close.return_value = None - return client - - -@pytest.fixture -def mock_binance_client(): - """Mock Binance client for testing.""" - client = AsyncMock() - - async def get_klines(symbol, interval, limit=500, start=None, end=None): - return [ - [0, 100.0 + i, 102.0 + i, 98.0 + i, 101.0 + i, 1500 + i] - for i in range(limit) - ] - - client.get_klines.side_effect = get_klines - client.get_price.return_value = 101.0 - client.close.return_value = None - return client - - @pytest.fixture def sample_portfolio_recommendation(): """Sample portfolio recommendation for testing.""" diff --git a/tests/unit/test_agents.py b/tests/unit/test_agents.py index 9c07634..6c168c2 100644 --- a/tests/unit/test_agents.py +++ b/tests/unit/test_agents.py @@ -4,7 +4,6 @@ from tradegraph_financial_advisor.agents.base_agent import BaseAgent from tradegraph_financial_advisor.agents.news_agent import NewsReaderAgent -from tradegraph_financial_advisor.agents.financial_agent import FinancialAnalysisAgent from tradegraph_financial_advisor.agents.report_analysis_agent import ( ReportAnalysisAgent, ) @@ -156,85 +155,6 @@ async def test_impact_score_calculation(self): assert 0.0 <= impact_score <= 1.0 assert impact_score > 0.5 # Should be higher due to symbol mention in title - - -class TestFinancialAnalysisAgent: - """Test FinancialAnalysisAgent functionality.""" - - @pytest.mark.asyncio - async def test_financial_agent_initialization( - self, mock_finnhub_client, mock_binance_client - ): - """Test financial agent initialization.""" - agent = FinancialAnalysisAgent( - finnhub_client=mock_finnhub_client, binance_client=mock_binance_client - ) - assert agent.name == "FinancialAnalysisAgent" - assert "financial" in agent.description.lower() - - @pytest.mark.asyncio - async def test_execute_financial_analysis( - self, mock_finnhub_client, mock_binance_client - ): - """Test financial analysis execution.""" - agent = FinancialAnalysisAgent( - finnhub_client=mock_finnhub_client, binance_client=mock_binance_client - ) - - await agent.start() - - input_data = { - "symbols": ["AAPL"], - "include_financials": True, - "include_technical": True, - "include_market_data": True, - } - - result = await agent.execute(input_data) - - assert "analysis_results" in result - assert "AAPL" in result["analysis_results"] - - aapl_data = result["analysis_results"]["AAPL"] - assert "market_data" in aapl_data - assert "financials" in aapl_data - assert "technical_indicators" in aapl_data - - await agent.stop() - - @pytest.mark.asyncio - async def test_market_data_extraction( - self, mock_finnhub_client, mock_binance_client - ): - """Test market data extraction.""" - agent = FinancialAnalysisAgent( - finnhub_client=mock_finnhub_client, binance_client=mock_binance_client - ) - - market_data = await agent._get_equity_market_data("AAPL") - - assert market_data is not None - assert market_data.symbol == "AAPL" - assert market_data.current_price > 0 - assert market_data.volume > 0 - - @pytest.mark.asyncio - async def test_technical_indicators_calculation( - self, mock_finnhub_client, mock_binance_client - ): - """Test technical indicators calculation.""" - agent = FinancialAnalysisAgent( - finnhub_client=mock_finnhub_client, binance_client=mock_binance_client - ) - - technical_data = await agent._get_equity_technical_indicators("AAPL") - - assert technical_data is not None - assert technical_data.symbol == "AAPL" - # Check that some indicators are calculated - assert technical_data.sma_20 is not None or technical_data.rsi is not None - - class TestReportAnalysisAgent: """Test ReportAnalysisAgent functionality.""" @@ -473,19 +393,3 @@ async def test_price_targets_calculation(self, mock_langchain_llm): assert target_price > current_price # Target should be higher for BUY if stop_loss: assert stop_loss < current_price # Stop loss should be lower for BUY - - -@pytest.mark.asyncio -async def test_agents_health_checks(mock_finnhub_client, mock_binance_client): - """Test health checks for all agents.""" - agents_to_test = [ - NewsReaderAgent(), - FinancialAnalysisAgent( - finnhub_client=mock_finnhub_client, binance_client=mock_binance_client - ), - ] - - for agent in agents_to_test: - health_ok = await agent.health_check() - # Health check should return a boolean - assert isinstance(health_ok, bool) From 891c2d26ad034c2bcbe50330a3804bf5b2e1e13e Mon Sep 17 00:00:00 2001 From: Mehran Moazeni Date: Wed, 10 Dec 2025 22:02:19 +0100 Subject: [PATCH 06/11] Allow workflow dependency injection --- .../workflows/analysis_workflow.py | 19 ++- tests/conftest.py | 25 ++++ tests/unit/test_workflows.py | 127 ++++++++++++------ 3 files changed, 122 insertions(+), 49 deletions(-) diff --git a/src/tradegraph_financial_advisor/workflows/analysis_workflow.py b/src/tradegraph_financial_advisor/workflows/analysis_workflow.py index eaea34c..f2207fe 100644 --- a/src/tradegraph_financial_advisor/workflows/analysis_workflow.py +++ b/src/tradegraph_financial_advisor/workflows/analysis_workflow.py @@ -40,18 +40,23 @@ def __init__( self, scraping_service: Optional[LocalScrapingService] = None, llm_model_name: str = "gpt-5-nano", + news_agent: Optional[NewsReaderAgent] = None, + financial_agent: Optional[FinancialAnalysisAgent] = None, + recommendation_engine: Optional[TradingRecommendationEngine] = None, + channel_service: Optional[FinancialNewsChannelService] = None, + llm: Optional[ChatOpenAI] = None, ): self.llm_model_name = llm_model_name - self.llm = ChatOpenAI( + self.llm = llm or ChatOpenAI( model=self.llm_model_name, temperature=0.1, api_key=settings.openai_api_key ) - self.news_agent = NewsReaderAgent() - self.financial_agent = FinancialAnalysisAgent() - self.recommendation_engine = TradingRecommendationEngine( + self.news_agent = news_agent or NewsReaderAgent() + self.financial_agent = financial_agent or FinancialAnalysisAgent() + self.recommendation_engine = recommendation_engine or TradingRecommendationEngine( model_name=self.llm_model_name ) self.local_scraping_service = scraping_service or LocalScrapingService() - self.channel_service = FinancialNewsChannelService() + self.channel_service = channel_service or FinancialNewsChannelService() self.workflow = None self._build_workflow() @@ -415,7 +420,7 @@ async def _generate_recommendations(self, state: AnalysisState) -> AnalysisState recommendations, portfolio_constraints, ) - state["recommendations"] = [rec.dict() for rec in optimized_recs] + state["recommendations"] = [rec.model_dump() for rec in optimized_recs] else: state["recommendations"] = [] @@ -957,7 +962,7 @@ def _article_to_dict(self, article: Any) -> Dict[str, Any]: if hasattr(article, "model_dump"): return article.model_dump() if hasattr(article, "dict"): - return article.dict() + return article.model_dump() return { "title": self._get_article_value(article, "title", ""), "url": self._get_article_value(article, "url", ""), diff --git a/tests/conftest.py b/tests/conftest.py index 5116302..857f30e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -411,3 +411,28 @@ def mock_local_scraping_service(): ] mock_service.health_check.return_value = True return mock_service + + +class _MockFinancialAgent: + """Simple mock financial agent for workflow tests.""" + + name = "FinancialAnalysisAgent" + description = "Mock financial agent" + + async def start(self): + return None + + async def stop(self): + return None + + async def execute(self, input_data): + return {"analysis_results": {}} + + async def health_check(self): + return True + + +@pytest.fixture +def mock_financial_agent(): + """Provide a lightweight financial agent replacement.""" + return _MockFinancialAgent() diff --git a/tests/unit/test_workflows.py b/tests/unit/test_workflows.py index 2896002..ceb6a64 100644 --- a/tests/unit/test_workflows.py +++ b/tests/unit/test_workflows.py @@ -11,11 +11,24 @@ class TestFinancialAnalysisWorkflow: """Test FinancialAnalysisWorkflow functionality.""" + def _create_workflow( + self, + mock_local_scraping_service, + mock_financial_agent, + ) -> FinancialAnalysisWorkflow: + """Helper to create workflows with mocked dependencies.""" + return FinancialAnalysisWorkflow( + scraping_service=mock_local_scraping_service, + financial_agent=mock_financial_agent, + ) + @pytest.mark.asyncio - async def test_workflow_initialization(self, mock_local_scraping_service): + async def test_workflow_initialization( + self, mock_local_scraping_service, mock_financial_agent + ): """Test workflow initialization.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) assert workflow.news_agent is not None @@ -24,10 +37,12 @@ async def test_workflow_initialization(self, mock_local_scraping_service): assert workflow.workflow is not None @pytest.mark.asyncio - async def test_analyze_portfolio_basic(self, mock_local_scraping_service): + async def test_analyze_portfolio_basic( + self, mock_local_scraping_service, mock_financial_agent + ): """Test basic portfolio analysis.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) workflow.llm = AsyncMock() workflow.llm.ainvoke.return_value = Mock( @@ -49,11 +64,14 @@ async def test_analyze_portfolio_basic(self, mock_local_scraping_service): @pytest.mark.asyncio async def test_collect_news_step( - self, mock_local_scraping_service, sample_news_articles + self, + mock_local_scraping_service, + mock_financial_agent, + sample_news_articles, ): """Test news collection workflow step.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) workflow.news_agent.execute = AsyncMock( return_value={ @@ -84,11 +102,14 @@ async def test_collect_news_step( @pytest.mark.asyncio async def test_analyze_financials_step( - self, mock_local_scraping_service, sample_financial_data + self, + mock_local_scraping_service, + mock_financial_agent, + sample_financial_data, ): """Test financial analysis workflow step.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) workflow.financial_agent.execute = AsyncMock( return_value={"analysis_results": sample_financial_data} @@ -116,11 +137,14 @@ async def test_analyze_financials_step( @pytest.mark.asyncio async def test_analyze_sentiment_step( - self, mock_local_scraping_service, sample_news_articles + self, + mock_local_scraping_service, + mock_financial_agent, + sample_news_articles, ): """Test sentiment analysis workflow step.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) workflow.llm = AsyncMock() workflow.llm.ainvoke.return_value = Mock( @@ -148,11 +172,14 @@ async def test_analyze_sentiment_step( @pytest.mark.asyncio async def test_generate_recommendations_step( - self, mock_local_scraping_service, sample_recommendations + self, + mock_local_scraping_service, + mock_financial_agent, + sample_recommendations, ): """Test recommendation generation workflow step.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) workflow.llm = AsyncMock() workflow.llm.ainvoke.return_value = Mock( @@ -199,11 +226,14 @@ async def test_generate_recommendations_step( @pytest.mark.asyncio async def test_create_portfolio_step( - self, mock_local_scraping_service, sample_recommendations + self, + mock_local_scraping_service, + mock_financial_agent, + sample_recommendations, ): """Test portfolio creation workflow step.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) workflow.llm = AsyncMock() workflow.llm.ainvoke.return_value = Mock( @@ -240,11 +270,14 @@ async def test_create_portfolio_step( @pytest.mark.asyncio async def test_validate_recommendations_step( - self, mock_local_scraping_service, sample_portfolio_recommendation + self, + mock_local_scraping_service, + mock_financial_agent, + sample_portfolio_recommendation, ): """Test recommendation validation workflow step.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) initial_state = AnalysisState( @@ -267,10 +300,12 @@ async def test_validate_recommendations_step( assert len(result_state["messages"]) > 0 @pytest.mark.asyncio - async def test_workflow_error_handling(self, mock_local_scraping_service): + async def test_workflow_error_handling( + self, mock_local_scraping_service, mock_financial_agent + ): """Test workflow error handling.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) workflow.news_agent.execute = AsyncMock(side_effect=Exception("Test error")) @@ -281,11 +316,11 @@ async def test_workflow_error_handling(self, mock_local_scraping_service): @pytest.mark.asyncio async def test_workflow_with_different_risk_tolerances( - self, mock_local_scraping_service + self, mock_local_scraping_service, mock_financial_agent ): """Test workflow with different risk tolerance settings.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) workflow.llm = AsyncMock() workflow.llm.ainvoke.return_value = Mock( @@ -309,10 +344,12 @@ async def test_workflow_with_different_risk_tolerances( assert result.get("portfolio_recommendation") is not None @pytest.mark.asyncio - async def test_workflow_state_transitions(self, mock_local_scraping_service): + async def test_workflow_state_transitions( + self, mock_local_scraping_service, mock_financial_agent + ): """Test workflow state transitions.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) # Test that workflow has proper state transitions @@ -322,10 +359,12 @@ async def test_workflow_state_transitions(self, mock_local_scraping_service): assert workflow.workflow is not None @pytest.mark.asyncio - async def test_workflow_with_multiple_symbols(self, mock_local_scraping_service): + async def test_workflow_with_multiple_symbols( + self, mock_local_scraping_service, mock_financial_agent + ): """Test workflow with multiple symbols.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) workflow.llm = AsyncMock() workflow.llm.ainvoke.return_value = Mock( @@ -366,10 +405,12 @@ async def test_workflow_with_multiple_symbols(self, mock_local_scraping_service) assert len(result["recommendations"]) >= 0 @pytest.mark.asyncio - async def test_workflow_performance(self, mock_local_scraping_service): + async def test_workflow_performance( + self, mock_local_scraping_service, mock_financial_agent + ): """Test workflow performance characteristics.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) workflow.llm = AsyncMock() workflow.llm.ainvoke.return_value = Mock( @@ -392,10 +433,12 @@ async def test_workflow_performance(self, mock_local_scraping_service): assert isinstance(result, dict) @pytest.mark.asyncio - async def test_workflow_cleanup(self, mock_local_scraping_service): + async def test_workflow_cleanup( + self, mock_local_scraping_service, mock_financial_agent + ): """Test workflow cleanup and resource management.""" - workflow = FinancialAnalysisWorkflow( - scraping_service=mock_local_scraping_service + workflow = self._create_workflow( + mock_local_scraping_service, mock_financial_agent ) workflow.news_agent = AsyncMock() workflow.financial_agent = AsyncMock() From 4ff5e206ba68228d75b89677c8c57917fb9a1917 Mon Sep 17 00:00:00 2001 From: Mehran Moazeni Date: Wed, 10 Dec 2025 22:05:58 +0100 Subject: [PATCH 07/11] Refresh settings for test API keys --- tests/conftest.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/tests/conftest.py b/tests/conftest.py index 857f30e..abdf949 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -5,6 +5,7 @@ from datetime import datetime, timedelta from tradegraph_financial_advisor.models.financial_data import NewsArticle +from tradegraph_financial_advisor.config.settings import refresh_openai_api_key # Test configuration os.environ["OPENAI_API_KEY"] = "test-openai-key" @@ -12,6 +13,9 @@ os.environ["FINNHUB_API_KEY"] = "test-finnhub-key" os.environ["LOG_LEVEL"] = "DEBUG" +# Ensure global settings pick up the test keys +refresh_openai_api_key() + @pytest.fixture(scope="session") def event_loop(): From 5afa1dedcf7a8235d34cc6a085459db0174b0ce3 Mon Sep 17 00:00:00 2001 From: Mehran Moazeni Date: Wed, 10 Dec 2025 22:08:38 +0100 Subject: [PATCH 08/11] Fix CI by setting Finnhub key --- .github/workflows/ci.yml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 25095a8..f399240 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,6 +34,8 @@ jobs: test: name: Unit Tests runs-on: ubuntu-latest + env: + FINNHUB_API_KEY: test-finnhub-key steps: - name: Checkout code uses: actions/checkout@v4 @@ -59,6 +61,8 @@ jobs: integration-tests: name: Integration Tests runs-on: ubuntu-latest + env: + FINNHUB_API_KEY: test-finnhub-key steps: - name: Checkout code uses: actions/checkout@v4 From c630b47a0742802522047692804b332398b260da Mon Sep 17 00:00:00 2001 From: Mehran Moazeni Date: Wed, 10 Dec 2025 22:12:23 +0100 Subject: [PATCH 09/11] Format codebase with Black --- examples/basic_usage.py | 4 +- .../agents/channel_report_agent.py | 14 +++- .../agents/financial_agent.py | 12 ++-- .../agents/multi_asset_allocation_agent.py | 64 +++++++------------ .../agents/news_agent.py | 2 +- .../agents/recommendation_engine.py | 9 +-- .../agents/report_analysis_agent.py | 4 +- .../config/settings.py | 37 +++++------ src/tradegraph_financial_advisor/main.py | 17 ++--- .../reporting/multi_asset_reporter.py | 4 +- .../reporting/pdf_reporter.py | 8 ++- .../services/channel_stream_service.py | 20 ++---- .../services/local_scraping_service.py | 4 +- .../services/price_trend_service.py | 8 ++- .../visualization/charts.py | 4 +- .../workflows/analysis_workflow.py | 16 ++--- tests/unit/test_agents.py | 2 + tests/unit/test_channels.py | 14 +++- tests/unit/test_models.py | 4 +- tests/unit/test_news_repository.py | 2 - 20 files changed, 125 insertions(+), 124 deletions(-) diff --git a/examples/basic_usage.py b/examples/basic_usage.py index 7832827..185846e 100644 --- a/examples/basic_usage.py +++ b/examples/basic_usage.py @@ -229,7 +229,9 @@ async def main(): # Check environment first if not check_environment(): - print("\n⚠️ Please configure your environment variables before running examples") + print( + "\n⚠️ Please configure your environment variables before running examples" + ) return # Run examples diff --git a/src/tradegraph_financial_advisor/agents/channel_report_agent.py b/src/tradegraph_financial_advisor/agents/channel_report_agent.py index 4bff574..b372818 100644 --- a/src/tradegraph_financial_advisor/agents/channel_report_agent.py +++ b/src/tradegraph_financial_advisor/agents/channel_report_agent.py @@ -70,7 +70,11 @@ async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: "price_action_notes (list), and key_stats (object describing counts)." ) response = await self.llm.ainvoke( - [HumanMessage(content=f"{prompt}\nINPUT:\n{json.dumps(prompt_payload)[:6000]}")] + [ + HumanMessage( + content=f"{prompt}\nINPUT:\n{json.dumps(prompt_payload)[:6000]}" + ) + ] ) data = json.loads(response.content) fallback.update({k: v for k, v in data.items() if v}) @@ -120,7 +124,9 @@ def _build_fallback_summary( trend_commentary = self._summarize_trends(price_trends) price_action_notes = self._build_price_notes(price_trends) - guidance_points = self._build_guidance_points(recommendations, price_action_notes) + guidance_points = self._build_guidance_points( + recommendations, price_action_notes + ) narrative_seed = " ".join(news_takeaways[:3]) or "Mixed market color." summary_text = ( @@ -158,7 +164,9 @@ def _build_key_stats( channel_payloads: Dict[str, Any], recommendations: List[Dict[str, Any]], ) -> Dict[str, Any]: - total_items = sum(len(payload.get("items", [])) for payload in channel_payloads.values()) + total_items = sum( + len(payload.get("items", [])) for payload in channel_payloads.values() + ) covered_symbols = { symbol for payload in channel_payloads.values() diff --git a/src/tradegraph_financial_advisor/agents/financial_agent.py b/src/tradegraph_financial_advisor/agents/financial_agent.py index 4762a70..aeae9ec 100644 --- a/src/tradegraph_financial_advisor/agents/financial_agent.py +++ b/src/tradegraph_financial_advisor/agents/financial_agent.py @@ -108,9 +108,7 @@ async def _get_equity_market_data(self, symbol: str) -> Optional[MarketData]: return None change = float(current_price) - float(open_price) - change_percent = ( - (change / float(open_price)) * 100 if open_price else 0.0 - ) + change_percent = (change / float(open_price)) * 100 if open_price else 0.0 volume = int(quote.get("v") or 0) market_cap = await self._get_market_cap(symbol) @@ -213,7 +211,9 @@ async def _get_equity_technical_indicators( high_series = pd.Series([float(value) for value in highs]) low_series = pd.Series([float(value) for value in lows]) - return self._build_technical_indicators(symbol, close_prices, high_series, low_series) + return self._build_technical_indicators( + symbol, close_prices, high_series, low_series + ) except Exception as e: logger.error( @@ -237,7 +237,9 @@ async def _get_crypto_technical_indicators( high_series = pd.Series([float(item[2]) for item in klines]) low_series = pd.Series([float(item[3]) for item in klines]) - return self._build_technical_indicators(symbol, close_prices, high_series, low_series) + return self._build_technical_indicators( + symbol, close_prices, high_series, low_series + ) except Exception as e: logger.error( diff --git a/src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py b/src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py index 51af66a..6e0da15 100644 --- a/src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py +++ b/src/tradegraph_financial_advisor/agents/multi_asset_allocation_agent.py @@ -49,7 +49,9 @@ def _build_strategy_library() -> Dict[str, Dict[str, Dict[str, AllocationSuggest ], } - def suggestion(asset_class: str, weight: float, rationale: str) -> AllocationSuggestion: + def suggestion( + asset_class: str, weight: float, rationale: str + ) -> AllocationSuggestion: return AllocationSuggestion( asset_class=asset_class, weight=weight, @@ -61,10 +63,14 @@ def suggestion(asset_class: str, weight: float, rationale: str) -> AllocationSug "growth": { "1w": { "stocks": suggestion( - "stocks", 0.35, "Stay liquid in quality mega caps while watching catalysts." + "stocks", + 0.35, + "Stay liquid in quality mega caps while watching catalysts.", ), "etfs": suggestion( - "etfs", 0.15, "Use QQQ/ARKK for beta exposure without security selection." + "etfs", + 0.15, + "Use QQQ/ARKK for beta exposure without security selection.", ), "crypto": suggestion( "crypto", 0.50, "Lean into BTC/ETH momentum for tactical upside." @@ -107,11 +113,11 @@ def suggestion(asset_class: str, weight: float, rationale: str) -> AllocationSug }, "1m": { "stocks": suggestion( - "stocks", 0.4, "Add cyclicals selectively as macro visibility improves." - ), - "etfs": suggestion( - "etfs", 0.4, "Core passive ETFs to anchor risk." + "stocks", + 0.4, + "Add cyclicals selectively as macro visibility improves.", ), + "etfs": suggestion("etfs", 0.4, "Core passive ETFs to anchor risk."), "crypto": suggestion( "crypto", 0.2, "Keep crypto beta but size for volatility." ), @@ -141,23 +147,15 @@ def suggestion(asset_class: str, weight: float, rationale: str) -> AllocationSug ), }, "1m": { - "stocks": suggestion( - "stocks", 0.3, "Income-oriented equities." - ), + "stocks": suggestion("stocks", 0.3, "Income-oriented equities."), "etfs": suggestion( "etfs", 0.55, "Blend IG bonds with broad equity ETFs." ), - "crypto": suggestion( - "crypto", 0.15, "Cap risk but allow for upside." - ), + "crypto": suggestion("crypto", 0.15, "Cap risk but allow for upside."), }, "1y": { - "stocks": suggestion( - "stocks", 0.35, "Quality and value tilt." - ), - "etfs": suggestion( - "etfs", 0.5, "VOO/TLT core plus dividend ETFs." - ), + "stocks": suggestion("stocks", 0.35, "Quality and value tilt."), + "etfs": suggestion("etfs", 0.5, "VOO/TLT core plus dividend ETFs."), "crypto": suggestion( "crypto", 0.15, "Long-dated call option sized exposure." ), @@ -171,31 +169,17 @@ def suggestion(asset_class: str, weight: float, rationale: str) -> AllocationSug "etfs": suggestion( "etfs", 0.55, "Covered-call and bond ETFs for yield." ), - "crypto": suggestion( - "crypto", 0.10, "Stablecoin yield or staking." - ), + "crypto": suggestion("crypto", 0.10, "Stablecoin yield or staking."), }, "1m": { - "stocks": suggestion( - "stocks", 0.4, "REITs + utilities blend." - ), - "etfs": suggestion( - "etfs", 0.5, "Bond ladders and dividend ETFs." - ), - "crypto": suggestion( - "crypto", 0.1, "Select staking strategies." - ), + "stocks": suggestion("stocks", 0.4, "REITs + utilities blend."), + "etfs": suggestion("etfs", 0.5, "Bond ladders and dividend ETFs."), + "crypto": suggestion("crypto", 0.1, "Select staking strategies."), }, "1y": { - "stocks": suggestion( - "stocks", 0.45, "Global dividend growth." - ), - "etfs": suggestion( - "etfs", 0.45, "Income ETFs and bond funds." - ), - "crypto": suggestion( - "crypto", 0.1, "Yield-focused crypto vehicles." - ), + "stocks": suggestion("stocks", 0.45, "Global dividend growth."), + "etfs": suggestion("etfs", 0.45, "Income ETFs and bond funds."), + "crypto": suggestion("crypto", 0.1, "Yield-focused crypto vehicles."), }, }, } diff --git a/src/tradegraph_financial_advisor/agents/news_agent.py b/src/tradegraph_financial_advisor/agents/news_agent.py index ebd1d15..7de2f48 100644 --- a/src/tradegraph_financial_advisor/agents/news_agent.py +++ b/src/tradegraph_financial_advisor/agents/news_agent.py @@ -54,7 +54,7 @@ async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: analyzed_articles = await self._analyze_articles(all_articles, symbols) return { - "articles": [article.dict() for article in analyzed_articles], + "articles": [article.model_dump() for article in analyzed_articles], "total_count": len(analyzed_articles), "sources": settings.news_sources, "analysis_timestamp": datetime.now().isoformat(), diff --git a/src/tradegraph_financial_advisor/agents/recommendation_engine.py b/src/tradegraph_financial_advisor/agents/recommendation_engine.py index f3d303e..8a3c734 100644 --- a/src/tradegraph_financial_advisor/agents/recommendation_engine.py +++ b/src/tradegraph_financial_advisor/agents/recommendation_engine.py @@ -62,11 +62,13 @@ async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]: ) return { - "individual_recommendations": [rec.dict() for rec in recommendations], + "individual_recommendations": [rec.model_dump() for rec in recommendations], "portfolio_recommendation": ( - portfolio_recommendation.dict() if portfolio_recommendation else None + portfolio_recommendation.model_dump() + if portfolio_recommendation + else None ), - "alerts": [alert.dict() for alert in alerts], + "alerts": [alert.model_dump() for alert in alerts], "generation_timestamp": datetime.now().isoformat(), } @@ -392,7 +394,6 @@ def _calculate_position_size( risk_preferences: Dict[str, Any], recommendation_type: RecommendationType, ) -> float: - RECOMMENDATION_WEIGHTS = { RecommendationType.STRONG_BUY: 2.0, RecommendationType.BUY: 1.5, diff --git a/src/tradegraph_financial_advisor/agents/report_analysis_agent.py b/src/tradegraph_financial_advisor/agents/report_analysis_agent.py index dfd86e9..421cb39 100644 --- a/src/tradegraph_financial_advisor/agents/report_analysis_agent.py +++ b/src/tradegraph_financial_advisor/agents/report_analysis_agent.py @@ -202,7 +202,9 @@ def _parse_json_response(self, payload: str) -> Dict[str, Any]: raise json.JSONDecodeError("Empty response", payload, 0) if "```" in cleaned: - segments = [segment.strip() for segment in cleaned.split("```") if segment.strip()] + segments = [ + segment.strip() for segment in cleaned.split("```") if segment.strip() + ] if segments: cleaned = segments[-1] diff --git a/src/tradegraph_financial_advisor/config/settings.py b/src/tradegraph_financial_advisor/config/settings.py index 1f3c0fa..5e9ad9a 100644 --- a/src/tradegraph_financial_advisor/config/settings.py +++ b/src/tradegraph_financial_advisor/config/settings.py @@ -3,20 +3,24 @@ from dotenv import load_dotenv from pydantic import Field -from pydantic_settings import BaseSettings +from pydantic_settings import BaseSettings, SettingsConfigDict load_dotenv() class Settings(BaseSettings): - openai_api_key: str = Field("", env="OPENAI_API_KEY") - finnhub_api_key: str = Field("", env="FINNHUB_API_KEY") - alpha_vantage_api_key: Optional[str] = Field(None, env="ALPHA_VANTAGE_API_KEY") - financial_data_api_key: Optional[str] = Field(None, env="FINANCIAL_DATA_API_KEY") + model_config = SettingsConfigDict( + env_file=".env", case_sensitive=False, extra="ignore" + ) + + openai_api_key: str = Field(default="") + finnhub_api_key: str = Field(default="") + alpha_vantage_api_key: Optional[str] = Field(default=None) + financial_data_api_key: Optional[str] = Field(default=None) - log_level: str = Field("INFO", env="LOG_LEVEL") - max_concurrent_agents: int = Field(5, env="MAX_CONCURRENT_AGENTS") - analysis_timeout_seconds: int = Field(30, env="ANALYSIS_TIMEOUT_SECONDS") + log_level: str = Field(default="INFO") + max_concurrent_agents: int = Field(default=5) + analysis_timeout_seconds: int = Field(default=30) news_sources: List[str] = Field( default_factory=lambda: [ @@ -25,20 +29,11 @@ class Settings(BaseSettings): "yahoo-finance", "marketwatch", "cnbc", - ], - env="NEWS_SOURCES", + ] ) - analysis_depth: str = Field("detailed", env="ANALYSIS_DEPTH") - default_portfolio_size: float = Field(100000.0, env="DEFAULT_PORTFOLIO_SIZE") - news_db_path: str = Field("tradegraph.duckdb", env="NEWS_DB_PATH") - - model_config = {"env_file": ".env", "case_sensitive": False, "extra": "ignore"} - - @classmethod - def get_news_sources_list(cls, v: str) -> List[str]: - if isinstance(v, str): - return [source.strip() for source in v.split(",")] - return v + analysis_depth: str = Field(default="detailed") + default_portfolio_size: float = Field(default=100000.0) + news_db_path: str = Field(default="tradegraph.duckdb") settings = Settings() diff --git a/src/tradegraph_financial_advisor/main.py b/src/tradegraph_financial_advisor/main.py index 24bd26d..3d14873 100644 --- a/src/tradegraph_financial_advisor/main.py +++ b/src/tradegraph_financial_advisor/main.py @@ -26,7 +26,9 @@ def __init__(self, llm_model_name: str = "gpt-5-nano"): model_name=self.llm_model_name ) self.report_analyzer = ReportAnalysisAgent(llm_model_name=self.llm_model_name) - self.channel_report_agent = ChannelReportAgent(llm_model_name=self.llm_model_name) + self.channel_report_agent = ChannelReportAgent( + llm_model_name=self.llm_model_name + ) self.channel_service = FinancialNewsChannelService() self.trend_service = PriceTrendService() self.pdf_report_writer = ChannelPDFReportWriter() @@ -167,7 +169,7 @@ async def quick_analysis( "analysis_type": "basic", "symbols": symbols, "recommendations": ( - [rec.dict() for rec in portfolio_rec.recommendations] + [rec.model_dump() for rec in portfolio_rec.recommendations] if portfolio_rec else [] ), @@ -614,8 +616,7 @@ async def main(): try: existing_reference = ( results - if isinstance(results, dict) - and results.get("channel_streams") + if isinstance(results, dict) and results.get("channel_streams") else None ) pdf_info = await advisor.generate_channel_pdf_report( @@ -627,13 +628,9 @@ async def main(): existing_results=existing_reference, output_path=args.pdf_path, ) - logger.info( - f"Channel PDF report saved to: {pdf_info['pdf_path']}" - ) + logger.info(f"Channel PDF report saved to: {pdf_info['pdf_path']}") except Exception as pdf_exc: - logger.warning( - f"Failed to create PDF channel report: {pdf_exc}" - ) + logger.warning(f"Failed to create PDF channel report: {pdf_exc}") # Display results based on output format if args.output_format == "json": diff --git a/src/tradegraph_financial_advisor/reporting/multi_asset_reporter.py b/src/tradegraph_financial_advisor/reporting/multi_asset_reporter.py index cde6dfe..e814ea9 100644 --- a/src/tradegraph_financial_advisor/reporting/multi_asset_reporter.py +++ b/src/tradegraph_financial_advisor/reporting/multi_asset_reporter.py @@ -45,9 +45,7 @@ def build_report( notes = plan.get("notes") or [] if notes: - cursor_y = self._draw_bullet_section( - doc, cursor_y, "Advisor Notes", notes - ) + cursor_y = self._draw_bullet_section(doc, cursor_y, "Advisor Notes", notes) for strategy in plan.get("strategies", []): cursor_y = self._draw_strategy_section(doc, cursor_y, strategy) diff --git a/src/tradegraph_financial_advisor/reporting/pdf_reporter.py b/src/tradegraph_financial_advisor/reporting/pdf_reporter.py index a57cd64..dccdc73 100644 --- a/src/tradegraph_financial_advisor/reporting/pdf_reporter.py +++ b/src/tradegraph_financial_advisor/reporting/pdf_reporter.py @@ -97,7 +97,9 @@ def build_report( f"Trend Notes: {summary_payload.get('trend_commentary', 'n/a')}", ) - cursor_y = self._draw_key_stats(doc, cursor_y, summary_payload.get("key_stats", {})) + cursor_y = self._draw_key_stats( + doc, cursor_y, summary_payload.get("key_stats", {}) + ) cursor_y = self._draw_channel_breakdown(doc, cursor_y, channel_payloads) cursor_y = self._draw_recommendations(doc, cursor_y, recommendations) @@ -191,7 +193,9 @@ def _draw_channel_breakdown( doc.drawString(self.margin, cursor_y, payload.get("title", channel_id)) cursor_y -= 14 doc.setFont("Helvetica", 10) - highlights = [item.get("title", "") for item in payload.get("items", [])[:3]] + highlights = [ + item.get("title", "") for item in payload.get("items", [])[:3] + ] body = "; ".join(highlights) or "No items collected" for line in self._wrap_text(body, 90): doc.drawString(self.margin + 10, cursor_y, f"- {line}") diff --git a/src/tradegraph_financial_advisor/services/channel_stream_service.py b/src/tradegraph_financial_advisor/services/channel_stream_service.py index 5f388f5..d2e2752 100644 --- a/src/tradegraph_financial_advisor/services/channel_stream_service.py +++ b/src/tradegraph_financial_advisor/services/channel_stream_service.py @@ -224,7 +224,9 @@ async def _get_session(self) -> aiohttp.ClientSession: return self._session def describe_channels(self) -> List[Dict[str, Any]]: - return [CHANNEL_REGISTRY[channel].metadata() for channel in self._iter_channels()] + return [ + CHANNEL_REGISTRY[channel].metadata() for channel in self._iter_channels() + ] async def collect_all_channels( self, symbols: Optional[Sequence[str]] = None @@ -238,9 +240,7 @@ async def collect_all_channels( result: Dict[str, Any] = {} for channel, payload in zip(self._iter_channels(), payloads): if isinstance(payload, Exception): - logger.warning( - f"Failed to collect channel {channel.value}: {payload}" - ) + logger.warning(f"Failed to collect channel {channel.value}: {payload}") continue result[channel.value] = payload return result @@ -291,9 +291,7 @@ async def _collect_news( collected: List[Dict[str, Any]] = [] for source, payload in zip(channel_definition.sources, results): if isinstance(payload, Exception): - logger.warning( - f"Failed to fetch feed from {source.name}: {payload}" - ) + logger.warning(f"Failed to fetch feed from {source.name}: {payload}") continue collected.extend(payload) @@ -358,9 +356,7 @@ def _normalize_entry( return None published = None - published_data = entry.get("published_parsed") or entry.get( - "updated_parsed" - ) + published_data = entry.get("published_parsed") or entry.get("updated_parsed") if published_data: published = datetime(*published_data[:6], tzinfo=timezone.utc) @@ -384,9 +380,7 @@ def _normalize_entry( "published_at": published.isoformat() if published else None, } - async def _build_price_items( - self, symbols: Sequence[str] - ) -> List[Dict[str, Any]]: + async def _build_price_items(self, symbols: Sequence[str]) -> List[Dict[str, Any]]: trends = await self.price_service.get_trends_for_symbols(list(symbols)) items: List[Dict[str, Any]] = [] for symbol in symbols: diff --git a/src/tradegraph_financial_advisor/services/local_scraping_service.py b/src/tradegraph_financial_advisor/services/local_scraping_service.py index 6141740..a1b924f 100644 --- a/src/tradegraph_financial_advisor/services/local_scraping_service.py +++ b/src/tradegraph_financial_advisor/services/local_scraping_service.py @@ -83,7 +83,9 @@ async def search_and_scrape_news( all_articles.append(article) new_articles.append(article) except Exception as scrape_exc: - logger.warning(f"Failed to scrape article {result.get('url')}: {scrape_exc}") + logger.warning( + f"Failed to scrape article {result.get('url')}: {scrape_exc}" + ) if new_articles: await self._persist_articles(new_articles) diff --git a/src/tradegraph_financial_advisor/services/price_trend_service.py b/src/tradegraph_financial_advisor/services/price_trend_service.py index 73838dc..90f9200 100644 --- a/src/tradegraph_financial_advisor/services/price_trend_service.py +++ b/src/tradegraph_financial_advisor/services/price_trend_service.py @@ -72,12 +72,16 @@ async def _run_symbol(self, symbol: str) -> Optional[Dict[str, Any]]: label: asyncio.create_task(self._fetch_trend(symbol, spec, now)) for label, spec in self._timeframes.items() } - results = await asyncio.gather(*trend_tasks.values(), return_exceptions=True) + results = await asyncio.gather( + *trend_tasks.values(), return_exceptions=True + ) trends: Dict[str, Any] = {} for label, result in zip(trend_tasks.keys(), results): if isinstance(result, Exception): - logger.warning(f"Trend window {label} failed for {symbol}: {result}") + logger.warning( + f"Trend window {label} failed for {symbol}: {result}" + ) continue if result: trends[label] = result diff --git a/src/tradegraph_financial_advisor/visualization/charts.py b/src/tradegraph_financial_advisor/visualization/charts.py index e22794a..2d86339 100644 --- a/src/tradegraph_financial_advisor/visualization/charts.py +++ b/src/tradegraph_financial_advisor/visualization/charts.py @@ -27,7 +27,9 @@ def create_portfolio_allocation_chart( portfolio_size = rec.get("portfolio_size") or 1 max_position = rec.get("max_position_size") allocation_value = ( - (max_position / portfolio_size) if max_position and portfolio_size else 0 + (max_position / portfolio_size) + if max_position and portfolio_size + else 0 ) allocations.append(float(allocation_value) * 100) diff --git a/src/tradegraph_financial_advisor/workflows/analysis_workflow.py b/src/tradegraph_financial_advisor/workflows/analysis_workflow.py index f2207fe..6d74da3 100644 --- a/src/tradegraph_financial_advisor/workflows/analysis_workflow.py +++ b/src/tradegraph_financial_advisor/workflows/analysis_workflow.py @@ -52,8 +52,9 @@ def __init__( ) self.news_agent = news_agent or NewsReaderAgent() self.financial_agent = financial_agent or FinancialAnalysisAgent() - self.recommendation_engine = recommendation_engine or TradingRecommendationEngine( - model_name=self.llm_model_name + self.recommendation_engine = ( + recommendation_engine + or TradingRecommendationEngine(model_name=self.llm_model_name) ) self.local_scraping_service = scraping_service or LocalScrapingService() self.channel_service = channel_service or FinancialNewsChannelService() @@ -89,7 +90,6 @@ async def analyze_portfolio( risk_tolerance: str = "medium", time_horizon: str = "medium_term", ) -> Dict[str, Any]: - if portfolio_size is None: portfolio_size = settings.default_portfolio_size @@ -190,9 +190,7 @@ async def _collect_news(self, state: AnalysisState) -> AnalysisState: ) state["channel_streams"] = channel_payloads except Exception as channel_exc: - logger.warning( - f"Failed to collect channel streams: {channel_exc}" - ) + logger.warning(f"Failed to collect channel streams: {channel_exc}") state["messages"].append( AIMessage(content=f"Collected {len(combined_news)} news articles") @@ -482,9 +480,9 @@ async def _create_portfolio(self, state: AnalysisState) -> AnalysisState: import json portfolio_data = json.loads(response.content) - portfolio_data["recommendations"] = ( - recommendations # Ensure recommendations are included - ) + portfolio_data[ + "recommendations" + ] = recommendations # Ensure recommendations are included state["portfolio_recommendation"] = portfolio_data except Exception as e: diff --git a/tests/unit/test_agents.py b/tests/unit/test_agents.py index 6c168c2..aa2b5bf 100644 --- a/tests/unit/test_agents.py +++ b/tests/unit/test_agents.py @@ -155,6 +155,8 @@ async def test_impact_score_calculation(self): assert 0.0 <= impact_score <= 1.0 assert impact_score > 0.5 # Should be higher due to symbol mention in title + + class TestReportAnalysisAgent: """Test ReportAnalysisAgent functionality.""" diff --git a/tests/unit/test_channels.py b/tests/unit/test_channels.py index a18970f..fe6fcfd 100644 --- a/tests/unit/test_channels.py +++ b/tests/unit/test_channels.py @@ -21,7 +21,9 @@ async def get_trends_for_symbols(self, symbols): @pytest.mark.asyncio async def test_channel_service_price_payload(sample_price_trends): - service = FinancialNewsChannelService(price_service=_DummyPriceService(sample_price_trends)) + service = FinancialNewsChannelService( + price_service=_DummyPriceService(sample_price_trends) + ) payload = await service.fetch_channel_payload( ChannelType.LIVE_PRICE_STREAM.value, symbols=["AAPL"] ) @@ -48,7 +50,9 @@ async def test_channel_report_agent_fallback( assert summary.get("key_stats", {}).get("channel_count") == 1 -def test_pdf_report_writer(tmp_path, sample_channel_streams, sample_price_trends, sample_recommendations): +def test_pdf_report_writer( + tmp_path, sample_channel_streams, sample_price_trends, sample_recommendations +): writer = ChannelPDFReportWriter() output_file = tmp_path / "report.pdf" summary_payload = { @@ -60,7 +64,11 @@ def test_pdf_report_writer(tmp_path, sample_channel_streams, sample_price_trends "advisor_memo": "Maintain constructive stance with risk controls.", "price_action_notes": ["AAPL: +2.5% weekly"], "guidance_points": ["Add MSFT on earnings strength"], - "key_stats": {"channel_count": 1, "headline_count": 2, "recommendation_count": 2}, + "key_stats": { + "channel_count": 1, + "headline_count": 2, + "recommendation_count": 2, + }, } pdf_path = writer.build_report( summary_payload=summary_payload, diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py index 56e0945..5ad6f43 100644 --- a/tests/unit/test_models.py +++ b/tests/unit/test_models.py @@ -312,7 +312,7 @@ def test_model_serialization(self): ) # Test dict conversion - rec_dict = recommendation.dict() + rec_dict = recommendation.model_dump() assert rec_dict["symbol"] == "AAPL" assert rec_dict["recommendation"] == "buy" assert rec_dict["risk_level"] == "medium" @@ -320,7 +320,7 @@ def test_model_serialization(self): # Test JSON serialization import json - json_str = recommendation.json() + json_str = recommendation.model_dump_json() parsed = json.loads(json_str) assert parsed["symbol"] == "AAPL" diff --git a/tests/unit/test_news_repository.py b/tests/unit/test_news_repository.py index a8c79c1..4146a8b 100644 --- a/tests/unit/test_news_repository.py +++ b/tests/unit/test_news_repository.py @@ -1,7 +1,5 @@ from datetime import datetime, timezone -import pytest - from tradegraph_financial_advisor.models.financial_data import NewsArticle from tradegraph_financial_advisor.repositories import NewsRepository From b77763f32203f1e0bb4a1db9eb3952852fbad166 Mon Sep 17 00:00:00 2001 From: Mehran Moazeni Date: Wed, 10 Dec 2025 22:16:18 +0100 Subject: [PATCH 10/11] Pin Black version for CI --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 073973d..0e1ccc7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,7 +53,7 @@ dev = [ "pytest>=7.4.0", "pytest-asyncio>=0.21.0", "pytest-cov>=4.1.0", - "black>=23.0.0", + "black==23.3.0", "isort>=5.12.0", "flake8>=6.0.0", "mypy>=1.5.0", From 9069a55ab5766018dcfc014371a1dac4b6379638 Mon Sep 17 00:00:00 2001 From: Mehran Moazeni Date: Mon, 15 Dec 2025 18:58:45 +0100 Subject: [PATCH 11/11] Fix linting on multi-asset branch --- src/tradegraph_financial_advisor/main.py | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/src/tradegraph_financial_advisor/main.py b/src/tradegraph_financial_advisor/main.py index f400628..2397a8b 100644 --- a/src/tradegraph_financial_advisor/main.py +++ b/src/tradegraph_financial_advisor/main.py @@ -194,14 +194,22 @@ async def quick_analysis( workflow_results, "sentiment_analysis", {} ) + portfolio_recommendation_payload = ( + portfolio_recommendation.dict() + if hasattr(portfolio_recommendation, "dict") + else portfolio_recommendation + ) + return { "analysis_type": "basic", "symbols": symbols, - "recommendations": ( - [rec.dict() for rec in portfolio_rec.recommendations] - if portfolio_rec - else [] + "recommendations": recommendations, + "portfolio_recommendation": ( + portfolio_recommendation_payload + if portfolio_recommendation_payload + else None ), + "sentiment_analysis": sentiment_analysis, "analysis_timestamp": datetime.now().isoformat(), }