#!/usr/bin/env python3 """ Bridge402 Earnings Transcript Client This example demonstrates how to retrieve earnings call transcripts from the Bridge402 Finance API using x402 payments. Features: - Retrieve earnings transcripts for public companies - Handle payment requirements - Error handling for invalid symbols/quarters - Batch processing for multiple companies Usage: python transcript_client.py --symbol AAPL --year 2024 --quarter 1 python transcript_client.py --batch-file companies.json """ import asyncio import json import base64 import logging import argparse from typing import Dict, Any, List, Optional from dataclasses import dataclass import httpx # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) @dataclass class TranscriptRequest: """Earnings transcript request parameters""" symbol: str year: int quarter: int class Bridge402TranscriptClient: """Client for retrieving earnings transcripts from Bridge402 Finance""" def __init__(self, api_url: str): self.api_url = api_url.rstrip('/') def create_mock_payment(self) -> str: """ Create a mock x402 payment for transcript access. In production, you would use the x402-python library to generate real EIP-3009 authorizations for $0.05 USDC per transcript. """ mock_payment = { "x402Version": 1, "scheme": "exact", "network": "base", "payload": { "signature": "0x" + "0" * 130, "authorization": "0x" + "0" * 130, } } payment_json = json.dumps(mock_payment) return base64.b64encode(payment_json.encode()).decode() async def get_transcript(self, request: TranscriptRequest) -> Optional[Dict[str, Any]]: """Retrieve a single earnings transcript""" logger.info(f"Getting {request.symbol} Q{request.quarter} {request.year} transcript...") # Validate parameters if request.quarter not in [1, 2, 3, 4]: logger.error(f"❌ Invalid quarter: {request.quarter}. Must be 1, 2, 3, or 4") return None if request.year < 2000 or request.year > 2030: logger.error(f"❌ Invalid year: {request.year}. Must be between 2000 and 2030") return None payment_data = self.create_mock_payment() headers = { "X-PAYMENT": payment_data, "Content-Type": "application/json" } async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.post( f"{self.api_url}/earnings-transcript", params={ "symbol": request.symbol.upper(), "year": request.year, "quarter": request.quarter }, headers=headers ) if response.status_code == 200: data = response.json() logger.info(f"✅ Transcript retrieved for {request.symbol}") return data elif response.status_code == 402: logger.warning("⚠️ Payment verification failed (expected with mock data)") return None elif response.status_code == 400: logger.error(f"❌ Invalid parameters: {response.text}") return None else: logger.error(f"❌ Request failed: {response.status_code} - {response.text}") return None except httpx.TimeoutException: logger.error("❌ Request timeout") return None except Exception as e: logger.error(f"❌ Request error: {e}") return None async def get_transcript_batch(self, requests: List[TranscriptRequest]) -> Dict[str, Any]: """Retrieve multiple transcripts in batch""" logger.info(f"Processing {len(requests)} transcript requests...") results = { "successful": [], "failed": [], "summary": { "total": len(requests), "successful_count": 0, "failed_count": 0 } } for request in requests: try: transcript = await self.get_transcript(request) if transcript: results["successful"].append({ "request": request, "transcript": transcript }) results["summary"]["successful_count"] += 1 else: results["failed"].append({ "request": request, "error": "Failed to retrieve transcript" }) results["summary"]["failed_count"] += 1 except Exception as e: results["failed"].append({ "request": request, "error": str(e) }) results["summary"]["failed_count"] += 1 # Small delay between requests to avoid rate limiting await asyncio.sleep(0.5) return results def format_transcript_summary(self, transcript: Dict[str, Any]) -> str: """Format transcript data for display""" symbol = transcript.get("symbol", "Unknown") year = transcript.get("year", "Unknown") quarter = transcript.get("quarter", "Unknown") transcript_data = transcript.get("transcript", {}) content = transcript_data.get("content", "") participants = transcript_data.get("participants", []) date = transcript_data.get("date", "Unknown") summary = f""" 📊 {symbol} Q{quarter} {year} Earnings Call Transcript {'=' * 50} Date: {date} Participants: {', '.join(participants) if participants else 'Not specified'} Content Preview: {content[:500]}{'...' if len(content) > 500 else ''} Payment Info: - Verified: {transcript.get('payment', {}).get('verified', False)} - Network: {transcript.get('payment', {}).get('network', 'Unknown')} - Transaction: {transcript.get('payment', {}).get('txHash', 'N/A')} Metadata: - Provider: {transcript.get('metadata', {}).get('provider', 'Unknown')} - Timestamp: {transcript.get('metadata', {}).get('timestamp', 'Unknown')} """ return summary def load_batch_requests(file_path: str) -> List[TranscriptRequest]: """Load transcript requests from JSON file""" try: with open(file_path, 'r') as f: data = json.load(f) requests = [] for item in data: request = TranscriptRequest( symbol=item["symbol"], year=item["year"], quarter=item["quarter"] ) requests.append(request) return requests except Exception as e: logger.error(f"❌ Error loading batch file: {e}") return [] async def demo_single_transcript(): """Demonstrate single transcript retrieval""" client = Bridge402TranscriptClient("http://localhost:8081") print("=" * 60) print("Single Transcript Demo") print("=" * 60) # Get AAPL Q1 2024 transcript request = TranscriptRequest(symbol="AAPL", year=2024, quarter=1) transcript = await client.get_transcript(request) if transcript: print(client.format_transcript_summary(transcript)) else: print("❌ Failed to retrieve transcript") async def demo_batch_transcripts(): """Demonstrate batch transcript retrieval""" client = Bridge402TranscriptClient("http://localhost:8081") print("\n" + "=" * 60) print("Batch Transcript Demo") print("=" * 60) # Create sample requests requests = [ TranscriptRequest(symbol="AAPL", year=2024, quarter=1), TranscriptRequest(symbol="MSFT", year=2024, quarter=1), TranscriptRequest(symbol="GOOGL", year=2024, quarter=1), TranscriptRequest(symbol="INVALID", year=2024, quarter=1), # This will fail ] results = await client.get_transcript_batch(requests) print(f"\n📊 Batch Results Summary:") print(f" Total requests: {results['summary']['total']}") print(f" Successful: {results['summary']['successful_count']}") print(f" Failed: {results['summary']['failed_count']}") if results["successful"]: print(f"\n✅ Successful transcripts:") for item in results["successful"]: request = item["request"] print(f" - {request.symbol} Q{request.quarter} {request.year}") if results["failed"]: print(f"\n❌ Failed requests:") for item in results["failed"]: request = item["request"] error = item["error"] print(f" - {request.symbol} Q{request.quarter} {request.year}: {error}") async def main(): """Main entry point""" parser = argparse.ArgumentParser(description="Bridge402 Transcript Client") parser.add_argument("--api-url", default="http://localhost:8081", help="Bridge402 API URL") parser.add_argument("--symbol", help="Stock symbol (e.g., AAPL)") parser.add_argument("--year", type=int, help="Year (e.g., 2024)") parser.add_argument("--quarter", type=int, help="Quarter (1, 2, 3, or 4)") parser.add_argument("--batch-file", help="JSON file with batch requests") parser.add_argument("--demo", action="store_true", help="Run demo scenarios") args = parser.parse_args() if args.demo: await demo_single_transcript() await demo_batch_transcripts() elif args.symbol and args.year and args.quarter: client = Bridge402TranscriptClient(args.api_url) request = TranscriptRequest(symbol=args.symbol, year=args.year, quarter=args.quarter) transcript = await client.get_transcript(request) if transcript: print(client.format_transcript_summary(transcript)) else: print("❌ Failed to retrieve transcript") elif args.batch_file: client = Bridge402TranscriptClient(args.api_url) requests = load_batch_requests(args.batch_file) if requests: results = await client.get_transcript_batch(requests) print(json.dumps(results, indent=2, default=str)) else: print("❌ No valid requests loaded") else: parser.print_help() if __name__ == "__main__": asyncio.run(main())