""" test_kaigora.py =============== Interactive manual tester for the Kaigora Agora Agent OpenAPI. Put this file in the same folder as kaigora_client.py, then: pip install requests export KAIGORA_API_KEY=ak_xxxxx # or it will prompt python test_kaigora.py Pick an endpoint from the menu. Every write action (POST) asks for y/N confirmation before actually hitting the API. """ from __future__ import annotations import json import os import sys from kaigora_client import KaigoraClient, KaigoraError # ------------------------------ utils ------------------------------ def pp(obj) -> None: """Pretty-print any JSON-ish object.""" print(json.dumps(obj, indent=2, ensure_ascii=False, default=str)) def ask(prompt: str, default=None, cast=str): """Prompt with optional default + type cast. Blank input -> default.""" suffix = f" [{default}]" if default is not None else "" s = input(f"{prompt}{suffix}: ").strip() if not s: return default try: return cast(s) except (ValueError, TypeError): print(f" ! invalid, using default ({default})") return default def confirm(msg: str) -> bool: return input(f"{msg} (y/N): ").strip().lower() == "y" # ------------------------------ test handlers ------------------------------ def t_participant_info(client: KaigoraClient) -> None: print("\n-- GET /my-participant-info --") pp(client.get_participant_info()) def t_portfolio(client: KaigoraClient) -> None: print("\n-- GET /my-portfolio --") pp(client.get_portfolio()) def t_order_window(client: KaigoraClient) -> None: print("\n-- orderWindow from /my-portfolio --") port = client.get_portfolio() pp(port.get("orderWindow")) print(f"\nisOpen = {(port.get('orderWindow') or {}).get('isOpen')}") def t_assets_page(client: KaigoraClient) -> None: print("\n-- GET /available-assets (single page) --") page = ask("page", 1, int) page_size = ask("pageSize (max 100)", 20, int) pp(client.get_available_assets(page=page, page_size=page_size)) def t_assets_all(client: KaigoraClient) -> None: print("\n-- GET /available-assets (all pages) --") assets = client.get_all_assets() print(f"Total: {len(assets)} assets\n") for a in assets[:30]: code = str(a.get("assetCode", "")) price = a.get("currentPrice") chg = a.get("change") print(f" {code:<8} {str(price):<12} {chg}") if len(assets) > 30: print(f" ... and {len(assets) - 30} more") def t_buy_amount(client: KaigoraClient) -> None: print("\n-- POST /orders (BUY by orderAmount) --") code = ask("assetCode", "AAPL") amount = ask("orderAmount (cash)", 1000, float) if confirm(f"Submit BUY {code} for ${amount}?"): pp(client.buy_by_amount(code, amount)) def t_buy_qty(client: KaigoraClient) -> None: print("\n-- POST /orders (BUY by orderQuantity) --") code = ask("assetCode", "AAPL") qty = ask("orderQuantity", 1, float) if confirm(f"Submit BUY {code} x {qty}?"): pp(client.buy_by_quantity(code, qty)) def t_sell(client: KaigoraClient) -> None: print("\n-- POST /orders (SELL) --") code = ask("assetCode", "AAPL") qty = ask("orderQuantity", 1, float) if confirm(f"Submit SELL {code} x {qty}?"): pp(client.sell(code, qty)) def t_batch(client: KaigoraClient) -> None: print("\n-- POST /orders (batch) --") print("Enter items one per line: SIDE ASSET amount|qty VALUE") print(" examples:") print(" BUY AAPL amount 1000") print(" BUY MSFT qty 3") print(" SELL AAPL qty 2") print("Blank line to finish.\n") items = [] while True: line = input("> ").strip() if not line: break parts = line.split() if len(parts) != 4: print(" ! format: SIDE ASSET amount|qty VALUE") continue side, code, kind, val = parts side = side.upper() if side not in ("BUY", "SELL"): print(" ! SIDE must be BUY or SELL") continue try: val = float(val) except ValueError: print(" ! VALUE must be a number") continue item = {"assetCode": code, "side": side} if kind.lower() == "amount": item["orderAmount"] = val elif kind.lower() == "qty": item["orderQuantity"] = val else: print(" ! kind must be 'amount' or 'qty'") continue items.append(item) if not items: print("No items entered.") return print("\nItems to submit:") pp(items) if confirm("Submit batch?"): pp(client.place_orders(items)) def t_cancel(client: KaigoraClient) -> None: print("\n-- POST /orders/{orderId}/cancel --") info = client.get_participant_info() pending = info.get("pendingOrders") or [] if not pending: print("No pending orders to cancel.") return print("Pending orders:") for i, o in enumerate(pending): print(f" [{i}] {o.get('orderId')} {o.get('side'):<4} " f"{o.get('assetCode'):<6} " f"amount={o.get('orderAmount')} qty={o.get('orderQuantity')}") idx = ask("Cancel which index? ('all' to cancel all)", 0) if str(idx).lower() == "all": if confirm(f"Cancel ALL {len(pending)} pending orders?"): pp(client.cancel_all_pending()) return try: idx = int(idx) except (TypeError, ValueError): print("! invalid index") return if 0 <= idx < len(pending): oid = pending[idx]["orderId"] if confirm(f"Cancel order {oid}?"): pp(client.cancel_order(oid)) else: print("! index out of range") # ------------------------------ menu loop ------------------------------ MENU = [ ("GET participant info", t_participant_info), ("GET portfolio (full)", t_portfolio), (" check orderWindow only", t_order_window), ("GET available assets (page)", t_assets_page), ("GET available assets (all)", t_assets_all), ("POST BUY by amount", t_buy_amount), ("POST BUY by quantity", t_buy_qty), ("POST SELL by quantity", t_sell), ("POST batch orders", t_batch), ("POST cancel pending order", t_cancel), ] def main() -> int: if not os.environ.get("KAIGORA_API_KEY"): key = input("KAIGORA_API_KEY not set. Paste API key: ").strip() if not key: print("No key provided, exiting.") return 1 os.environ["KAIGORA_API_KEY"] = key client = KaigoraClient() while True: print("\n" + "=" * 52) print(" Kaigora API manual tester") print("=" * 52) for i, (label, _) in enumerate(MENU): print(f" [{i}] {label}") print(" [q] quit") choice = input("\n> ").strip().lower() if choice in ("q", "quit", "exit"): return 0 if not choice.isdigit() or int(choice) >= len(MENU): print("! invalid choice") continue _, fn = MENU[int(choice)] try: fn(client) except KaigoraError as e: print(f"\n[KaigoraError] {type(e).__name__}: {e}") except KeyboardInterrupt: print("\n(cancelled)") except Exception as e: print(f"\n[unhandled {type(e).__name__}] {e}") if __name__ == "__main__": sys.exit(main() or 0)