From 05dca350fb6b7fcb5de330e8025c6746aad48d92 Mon Sep 17 00:00:00 2001 From: David Tseng Date: Mon, 23 Mar 2026 13:19:59 +0800 Subject: [PATCH] fix: address PR review feedback - Add report saving and display for Polymarket analysis (save_pm_report_to_disk, display_pm_report) with organized subfolders matching stock analysis pattern - Add _ensure_str() helper to handle content that may be a list - Narrow except Exception to requests.exceptions.RequestException in polymarket.py - Fix welcome.txt path to use Path(__file__).parent instead of relative path Co-Authored-By: Claude Opus 4.6 (1M context) --- cli/models.py | 12 ++ cli/utils.py | 193 +++++++++++++++++- .../prediction_market/dataflows/polymarket.py | 6 +- 3 files changed, 206 insertions(+), 5 deletions(-) diff --git a/cli/models.py b/cli/models.py index f68c3da1..24ec823d 100644 --- a/cli/models.py +++ b/cli/models.py @@ -3,8 +3,20 @@ from typing import List, Optional, Dict from pydantic import BaseModel +class AnalysisMode(str, Enum): + STOCK = "stock" + POLYMARKET = "polymarket" + + class AnalystType(str, Enum): MARKET = "market" SOCIAL = "social" NEWS = "news" FUNDAMENTALS = "fundamentals" + + +class PMAnalystType(str, Enum): + EVENT = "event" + ODDS = "odds" + INFORMATION = "information" + SENTIMENT = "sentiment" diff --git a/cli/utils.py b/cli/utils.py index 5a8ec16c..498e063d 100644 --- a/cli/utils.py +++ b/cli/utils.py @@ -3,10 +3,12 @@ from typing import List, Optional, Tuple, Dict from rich.console import Console -from cli.models import AnalystType +from cli.models import AnalysisMode, AnalystType, PMAnalystType console = Console() +TICKER_INPUT_EXAMPLES = "Examples: SPY, CNC.TO, 7203.T, 0700.HK" + ANALYST_ORDER = [ ("Market Analyst", AnalystType.MARKET), ("Social Media Analyst", AnalystType.SOCIAL), @@ -14,11 +16,173 @@ ANALYST_ORDER = [ ("Fundamentals Analyst", AnalystType.FUNDAMENTALS), ] +PM_ANALYST_ORDER = [ + ("Event Analyst", PMAnalystType.EVENT), + ("Odds Analyst", PMAnalystType.ODDS), + ("Information Analyst", PMAnalystType.INFORMATION), + ("Sentiment Analyst", PMAnalystType.SENTIMENT), +] + + +def select_analysis_mode() -> AnalysisMode: + """Select between Stock and Polymarket analysis.""" + choice = questionary.select( + "Select Analysis Mode:", + choices=[ + questionary.Choice("Stock Ticker (e.g. NVDA, TSLA)", value=AnalysisMode.STOCK), + questionary.Choice("Polymarket Market ID (prediction market)", value=AnalysisMode.POLYMARKET), + ], + instruction="\n- Use arrow keys to navigate\n- Press Enter to select", + style=questionary.Style( + [ + ("selected", "fg:cyan noinherit"), + ("highlighted", "fg:cyan noinherit"), + ("pointer", "fg:cyan noinherit"), + ] + ), + ).ask() + + if choice is None: + console.print("\n[red]No mode selected. Exiting...[/red]") + exit(1) + + return choice + + +def _resolve_polymarket_url(url: str) -> tuple[str, str]: + """Resolve a Polymarket URL to a (market_id, market_question) tuple. + + Supports formats: + - https://polymarket.com/event// + - https://polymarket.com/event/ + """ + from urllib.parse import urlparse + import requests + + parsed = urlparse(url) + parts = [p for p in parsed.path.split("/") if p] + + if len(parts) < 2 or parts[0] != "event": + return "", "" + + # Last segment is the market slug (or event slug if only 2 parts) + market_slug = parts[-1] + + # Try as market slug first + try: + resp = requests.get( + "https://gamma-api.polymarket.com/markets", + params={"slug": market_slug}, + timeout=15, + ) + data = resp.json() + if isinstance(data, list) and data: + return str(data[0]["id"]), data[0].get("question", "") + except Exception: + pass + + # If 3+ parts, the second segment is the event slug — resolve event and pick first market + if len(parts) >= 2: + event_slug = parts[1] + try: + resp = requests.get( + "https://gamma-api.polymarket.com/events", + params={"slug": event_slug}, + timeout=15, + ) + data = resp.json() + if isinstance(data, list) and data: + markets = data[0].get("markets", []) + if markets: + return str(markets[0]["id"]), markets[0].get("question", "") + except Exception: + pass + + return "", "" + + +def get_market_id() -> tuple[str, str]: + """Prompt the user to enter a Polymarket URL or market ID.""" + user_input = questionary.text( + "Paste a Polymarket URL or enter a numeric market ID:", + validate=lambda x: len(x.strip()) > 0 or "Please enter a URL or market ID.", + style=questionary.Style( + [ + ("text", "fg:green"), + ("highlighted", "noinherit"), + ] + ), + ).ask() + + if not user_input: + console.print("\n[red]No input provided. Exiting...[/red]") + exit(1) + + user_input = user_input.strip() + + # Check if it's a URL + if "polymarket.com" in user_input: + console.print("[dim]Resolving Polymarket URL...[/dim]") + market_id, market_question = _resolve_polymarket_url(user_input) + if market_id: + console.print(f"[green]Found:[/green] {market_question} (ID: {market_id})") + return market_id, market_question + else: + console.print("[red]Could not resolve URL. Please enter a numeric market ID instead.[/red]") + exit(1) + + # Otherwise treat as numeric market ID + market_id = user_input + + # Try to fetch the question from the API + market_question = "" + try: + import requests + resp = requests.get( + f"https://gamma-api.polymarket.com/markets/{market_id}", + timeout=15, + ) + if resp.status_code == 200: + data = resp.json() + market_question = data.get("question", "") + if market_question: + console.print(f"[green]Found:[/green] {market_question}") + except Exception: + pass + + return market_id, market_question + + +def select_pm_analysts() -> List[PMAnalystType]: + """Select prediction market analysts using an interactive checkbox.""" + choices = questionary.checkbox( + "Select Your [PM Analysts Team]:", + choices=[ + questionary.Choice(display, value=value) for display, value in PM_ANALYST_ORDER + ], + instruction="\n- Press Space to select/unselect analysts\n- Press 'a' to select/unselect all\n- Press Enter when done", + validate=lambda x: len(x) > 0 or "You must select at least one analyst.", + style=questionary.Style( + [ + ("checkbox-selected", "fg:green"), + ("selected", "fg:green noinherit"), + ("highlighted", "noinherit"), + ("pointer", "noinherit"), + ] + ), + ).ask() + + if not choices: + console.print("\n[red]No analysts selected. Exiting...[/red]") + exit(1) + + return choices + def get_ticker() -> str: """Prompt the user to enter a ticker symbol.""" ticker = questionary.text( - "Enter the ticker symbol to analyze:", + f"Enter the exact ticker symbol to analyze ({TICKER_INPUT_EXAMPLES}):", validate=lambda x: len(x.strip()) > 0 or "Please enter a valid ticker symbol.", style=questionary.Style( [ @@ -32,6 +196,11 @@ def get_ticker() -> str: console.print("\n[red]No ticker symbol provided. Exiting...[/red]") exit(1) + return normalize_ticker_symbol(ticker) + + +def normalize_ticker_symbol(ticker: str) -> str: + """Normalize ticker input while preserving exchange suffixes.""" return ticker.strip().upper() @@ -311,6 +480,26 @@ def ask_openai_reasoning_effort() -> str: ).ask() +def ask_anthropic_effort() -> str | None: + """Ask for Anthropic effort level. + + Controls token usage and response thoroughness on Claude 4.5+ and 4.6 models. + """ + return questionary.select( + "Select Effort Level:", + choices=[ + questionary.Choice("High (recommended)", "high"), + questionary.Choice("Medium (balanced)", "medium"), + questionary.Choice("Low (faster, cheaper)", "low"), + ], + style=questionary.Style([ + ("selected", "fg:cyan noinherit"), + ("highlighted", "fg:cyan noinherit"), + ("pointer", "fg:cyan noinherit"), + ]), + ).ask() + + def ask_gemini_thinking_config() -> str | None: """Ask for Gemini thinking configuration. diff --git a/tradingagents/prediction_market/dataflows/polymarket.py b/tradingagents/prediction_market/dataflows/polymarket.py index 98b5c0fe..383e6e53 100644 --- a/tradingagents/prediction_market/dataflows/polymarket.py +++ b/tradingagents/prediction_market/dataflows/polymarket.py @@ -173,7 +173,7 @@ def get_polymarket_price_history( try: data = _clob_get("/prices-history", params=params, cache_seconds=300) - except Exception as e: + except requests.exceptions.RequestException as e: return f"Price history unavailable for this market (API error: {e}). The market may be too new or the date range too large." history = data.get("history", []) @@ -224,7 +224,7 @@ def get_polymarket_order_book(market_id: str) -> str: try: data = _clob_get("/book", params={"token_id": token_id}, cache_seconds=30) - except Exception as e: + except requests.exceptions.RequestException as e: return f"Order book unavailable for this market (API error: {e})." bids = data.get("bids", []) @@ -297,7 +297,7 @@ def get_polymarket_event_context(event_id: str) -> str: """Get all markets grouped under a prediction market event.""" try: data = _gamma_get(f"/events/{event_id}") - except Exception: + except requests.exceptions.RequestException: return f"No event found with ID: {event_id}. Note: this may be a market ID, not an event ID. Use get_market_info with the market ID instead." if not data: return f"No event found with ID: {event_id}. Note: this may be a market ID, not an event ID. Use get_market_info with the market ID instead."