#!/usr/bin/env python3 """ LuckySt Syndicate - Report & Schedule API Endpoints FastAPI router for: - Listing and reading encrypted on-chain trading reports - Scheduling games with Base on-chain commitment """ import os from fastapi import APIRouter, HTTPException from pydantic import BaseModel from typing import Optional import redis from agentic.syndicate.report_crypto import ReportKeyManager, EncryptedReport, decrypt_report from agentic.syndicate.chain_writer import fetch_tx_calldata from agentic.syndicate.game_scheduler import ( create_schedule, list_pending_games, list_active_games, ) router = APIRouter(tags=["syndicate-reports"]) def _get_redis(): """Get Redis client from environment.""" return redis.Redis( host=os.environ.get("REDIS_HOST", "localhost"), port=int(os.environ.get("REDIS_PORT", 6379)), password=os.environ.get("REDIS_PASSWORD"), db=int(os.environ.get("REDIS_DB", 0)), ) # --------------------------------------------------------------------------- # Schemas # --------------------------------------------------------------------------- class AccessRequest(BaseModel): tx_hash: str requester_address: str author_address: str class AccessResponse(BaseModel): aes_key_hex: Optional[str] = None error: Optional[str] = None class ReportListItem(BaseModel): tx_hash: str author: str instance_id: int markets: list[str] timestamp: str class DecryptedReport(BaseModel): tx_hash: str plaintext: str # --------------------------------------------------------------------------- # Endpoints # --------------------------------------------------------------------------- @router.get("/reports/by-author/{address}", response_model=list[ReportListItem]) async def list_reports_by_author(address: str): """List all report tx hashes for a given author address.""" r = _get_redis() manager = ReportKeyManager(r) reports = manager.list_reports_by_author(address) return [ ReportListItem( tx_hash=rpt.get("tx_hash", ""), author=rpt.get("author", ""), instance_id=rpt.get("instance_id", 0), markets=rpt.get("markets", []), timestamp=rpt.get("timestamp", ""), ) for rpt in reports ] @router.post("/reports/request-access", response_model=AccessResponse) async def request_report_access(req: AccessRequest): """ Request a 24-hour decryption key for a report. The requester must be a pledged syndicate member (checked against registry.yaml). Authors always get access to their own reports. """ r = _get_redis() manager = ReportKeyManager(r) key_hex = manager.grant_access( tx_hash=req.tx_hash, requester_address=req.requester_address, author_address=req.author_address, ) if key_hex: return AccessResponse(aes_key_hex=key_hex) else: return AccessResponse(error="Access denied. Requester must be a pledged syndicate member.") @router.post("/reports/decrypt", response_model=DecryptedReport) async def decrypt_onchain_report(req: AccessRequest): """ Full read flow: request access key, fetch calldata from Base, decrypt. Convenience endpoint that combines access request + fetch + decrypt. """ r = _get_redis() manager = ReportKeyManager(r) # Get the AES key key_hex = manager.grant_access( tx_hash=req.tx_hash, requester_address=req.requester_address, author_address=req.author_address, ) if not key_hex: raise HTTPException(status_code=403, detail="Access denied. Not a pledged syndicate member.") # Fetch calldata from Base raw_data = await fetch_tx_calldata(req.tx_hash) if not raw_data: raise HTTPException(status_code=404, detail="Transaction not found or has no calldata.") # Parse and decrypt try: version, nonce, ciphertext = EncryptedReport.parse_chain_bytes(raw_data) key_bytes = bytes.fromhex(key_hex) plaintext = decrypt_report(ciphertext, nonce, key_bytes) return DecryptedReport(tx_hash=req.tx_hash, plaintext=plaintext) except Exception as e: raise HTTPException(status_code=500, detail=f"Decryption failed: {e}") # --------------------------------------------------------------------------- # Game Schedule Schemas # --------------------------------------------------------------------------- class GameEntrySchema(BaseModel): markets: list[str] start_time: str # ISO-8601 UTC stop_time: Optional[str] = None contract_size: int = 1 higher_first: bool = False class ScheduleRequest(BaseModel): games: list[GameEntrySchema] class ScheduleResponse(BaseModel): schedule_id: str game_count: int base_tx: Optional[str] = None class ScheduleListItem(BaseModel): schedule_id: str game_index: int markets: list[str] start_time: str stop_time: Optional[str] = None status: str instance_id: Optional[int] = None base_tx: Optional[str] = None # --------------------------------------------------------------------------- # Game Schedule Endpoints # --------------------------------------------------------------------------- @router.post("/schedule/games", response_model=ScheduleResponse) async def schedule_games(req: ScheduleRequest): """ Schedule a list of games for automated trading. Each game specifies market tickers, start time, and optional stop time. The schedule is committed to Base as an immutable on-chain record, then stored in Redis for the execution engine to deploy at game time. """ if not req.games: raise HTTPException(status_code=400, detail="No games provided.") if len(req.games) > 20: raise HTTPException(status_code=400, detail="Maximum 20 games per schedule.") r = _get_redis() base_address = os.environ.get("LUCKBOT_BASE_ADDRESS") base_key = os.environ.get("LUCKBOT_BASE_PRIVATE_KEY") games_dicts = [g.model_dump() for g in req.games] commitment = await create_schedule( games=games_dicts, redis_client=r, base_address=base_address, base_key=base_key, ) return ScheduleResponse( schedule_id=commitment.schedule_id, game_count=len(commitment.games), base_tx=commitment.base_tx, ) @router.get("/schedule/pending", response_model=list[ScheduleListItem]) async def get_pending_schedule(): """List all pending scheduled games (not yet deployed).""" r = _get_redis() games = list_pending_games(r) return [ ScheduleListItem( schedule_id=g.get("schedule_id", ""), game_index=g.get("game_index", 0), markets=g.get("markets", []), start_time=g.get("start_time", ""), stop_time=g.get("stop_time"), status=g.get("status", "pending"), base_tx=g.get("base_tx"), ) for g in games ] @router.get("/schedule/active", response_model=list[ScheduleListItem]) async def get_active_schedule(): """List all currently running scheduled games.""" r = _get_redis() games = list_active_games(r) return [ ScheduleListItem( schedule_id=g.get("schedule_id", ""), game_index=g.get("game_index", 0), markets=g.get("markets", []), start_time=g.get("start_time", ""), stop_time=g.get("stop_time"), status=g.get("status", "running"), instance_id=g.get("instance_id"), base_tx=g.get("base_tx"), ) for g in games ]