#!/usr/bin/env python3 """SYGNL/47 starter agent. Standard library only, no dependencies. curl -O https://47.sygnliq.com/agent.py python3 agent.py --help python3 agent.py --manifest python3 agent.py --tier archive --once --dry-run WHAT YOU ARE BUYING, AND WHAT YOU ARE NOT ========================================= SYGNL/47 sells an OBSERVATION FEED. Each event says: at this UTC second, a US president named this public company on broadcast audio, in this quoted context, the audio was checked live or replay, this many independent feeds heard it, and the instrument moved this much against its benchmark afterwards. Every one of those fields is a checkable fact about something that already happened. It is not a strategy. It is not a signal with a track record. Read this before you wire it to anything: * ZERO CONFIRMED LIVE OUTCOMES. Nothing in this feed has ever been observed live and then traded. Not once. * n = 2. Two events are on file. Both are RECONSTRUCTED FROM ARCHIVED BROADCAST AUDIO, not live captures, and both come from the same speech, the same stock and the same day. That is not two observations. It is closer to one. * The publisher's own event study returns INSUFFICIENT at this sample size and will keep returning it below twenty independent events, however good the numbers look. * Price moves in the feed are measured on bar closes. They are marks, not fills. Nobody has measured what you would actually get filled at. * There is no win rate here, no backtest, no edge claim, and no profit claim. If you find one in this file, it is a bug. So: NOTHING ABOUT THIS SIGNAL IS EVIDENCED. If you connect the execution path below to real money, you are trading on an unevidenced hypothesis with your own capital, and the outcome is entirely your risk. The honest use of this file is to observe, log, and build your own evidence before risking anything. EXECUTION SAFETY ================ Execution is OFF unless you pass --execute. Even then: * PAPER by default, always. Live requires TWO deliberate environment variables, mirroring the publisher's own broker discipline: export SYGNL47_BROKER_MODE=live export SYGNL47_LIVE_CONFIRM=I_UNDERSTAND_THIS_TRADES_REAL_MONEY One stray variable is not enough to move from simulated fills to real money. * LIMIT ORDERS ONLY. This file contains no market-order path. A market order into a headline is the fastest way to turn a paper result into a real loss. * A per-fire notional cap and a hard daily notional cap, both enforced before an order is built, plus a cap on fires per day. * Reconstructed (source: replay) events are never executed. Neither are events whose liveness check did not come back LIVE. CREDENTIALS =========== None are in this file and none should ever be written into it. Everything comes from the environment: export SYGNL47_APCA_KEY=... # broker key (Alpaca, paper to start) export SYGNL47_APCA_SECRET=... # broker secret export SYGNL47_PASS=... # optional SYGNL/47 subscription pass export SYGNL47_PAYMENT_SIGNATURE=... # optional per-call x402 payload PAYING FOR THE FEED =================== Priced routes answer HTTP 402 with x402 payment requirements: USDC on Base mainnet. Two ways to satisfy them, both handled here. 1. A subscription pass. Buy one once at POST https://47.sygnliq.com/v1/potus/subscribe/live/24h (or live/30d, or archive/30d), keep the token it returns in SYGNL47_PASS, and this agent sends it as X-SYGNL47-TOKEN on every poll. Simplest path for a long poll. 2. Per call. Sign the EIP-3009 authorization the 402 body asks for with your own wallet tooling, put the base64 PaymentPayload in SYGNL47_PAYMENT_SIGNATURE, and it is sent as PAYMENT-SIGNATURE. Legacy x402 v1 clients that send X-PAYMENT are also accepted by the seller. With neither set, this agent still runs: it prints the exact payment requirements from the 402 body and keeps polling. You are never charged for an empty window (404) or a stale artifact (503); neither is settled. """ from __future__ import annotations import argparse import json import os import sys import time import urllib.error import urllib.request from datetime import datetime, timezone BASE = os.environ.get("SYGNL47_BASE", "https://47.sygnliq.com") TIER_ROUTES = { "live": "/v1/potus/live", "1m": "/v1/potus/events/1m", "15m": "/v1/potus/events/15m", "archive": "/v1/potus/archive", } TIER_PRICE_USD = {"live": 2.00, "1m": 0.75, "15m": 0.25, "archive": 0.10} PASS_HEADER = "X-SYGNL47-TOKEN" PAY_HEADER = "PAYMENT-SIGNATURE" # --- execution caps. Deliberately small. Raise them consciously, never by # --- accident, and never before you have your own evidence. PER_FIRE_USD = float(os.environ.get("SYGNL47_PER_FIRE_USD", "500")) DAILY_NOTIONAL_USD = float(os.environ.get("SYGNL47_DAILY_NOTIONAL_USD", "1000")) MAX_FIRES_PER_DAY = int(os.environ.get("SYGNL47_MAX_FIRES_PER_DAY", "2")) LIMIT_OFFSET_BPS = float(os.environ.get("SYGNL47_LIMIT_OFFSET_BPS", "15")) PAPER_URL = "https://paper-api.alpaca.markets" LIVE_URL = "https://api.alpaca.markets" DATA_URL = "https://data.alpaca.markets" DISCLAIMER = ( "SYGNL/47 is an observation feed. Zero confirmed live outcomes. " "Two events on file, both reconstructed from archived broadcast audio. " "Not investment advice, not a recommendation, not a claim of predictive " "value. Nothing here is evidenced." ) def log(msg: str) -> None: stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") print(f"{stamp} {msg}", flush=True) # ============================================================== feed client class Feed: """Reads the SYGNL/47 event ladder. Handles the 402 and the retry.""" def __init__(self, base: str = BASE, timeout: int = 20): self.base = base.rstrip("/") self.timeout = timeout self.sub_pass = os.environ.get("SYGNL47_PASS", "").strip() self.payment = os.environ.get("SYGNL47_PAYMENT_SIGNATURE", "").strip() def credentialled(self) -> bool: return bool(self.sub_pass or self.payment) def _headers(self, paid: bool) -> dict: h = {"Accept": "application/json", "User-Agent": "sygnl47-starter/1.0"} if paid and self.sub_pass: h[PASS_HEADER] = self.sub_pass if paid and self.payment: h[PAY_HEADER] = self.payment return h def _get(self, path: str, paid: bool) -> tuple[int, dict]: url = f"{self.base}{path}" req = urllib.request.Request(url, headers=self._headers(paid), method="GET") try: with urllib.request.urlopen(req, timeout=self.timeout) as r: return r.status, json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read() or b"{}") except (ValueError, OSError): return e.code, {} except Exception as e: # noqa: BLE001 return 0, {"error": f"{type(e).__name__}: {e}"} def manifest(self) -> dict: """The free service manifest. No payment, no events, honest inventory.""" code, body = self._get("/v1/potus", paid=False) if code != 200: return {"error": f"manifest unavailable (http {code})"} return body def poll(self, tier: str, limit: int = 100) -> tuple[str, dict]: """(state, payload). state is one of: ok events returned empty nothing inside this tier's window, 404, NOT charged unavailable publisher sync missing, 503, NOT charged payment 402, and we had nothing to pay with error anything else """ route = f"{TIER_ROUTES[tier]}?limit={int(limit)}" # First call is unpaid on purpose when we hold no credential: it costs # nothing and it returns the exact requirements to satisfy. code, body = self._get(route, paid=self.credentialled()) if code == 402: if not self.credentialled(): return "payment", body # A pass can expire mid-run. The seller then falls back to a normal # 402 rather than serving free, so report it plainly. return "payment", body if code == 200: return "ok", body if code == 404: return "empty", body if code == 503: return "unavailable", body return "error", {"http": code, **(body if isinstance(body, dict) else {})} def describe_402(body: dict) -> str: """Turn an x402 challenge into one readable line.""" accepts = (body or {}).get("accepts") or [] if not accepts: return "402 payment required, no accepts block in the response" a = accepts[0] raw = str(a.get("amount") or "") usd = "" if raw.isdigit(): usd = f" (${int(raw) / 1_000_000:.2f} USDC)" return (f"402 payment required: {raw} base units{usd} on {a.get('network')} " f"asset {a.get('asset')} to {a.get('payTo')}. " f"Sign the EIP-3009 authorization and retry with the " f"{PAY_HEADER} header, or buy a pass and set SYGNL47_PASS.") # ================================================================== events def event_line(ev: dict) -> str: v = ev.get("verification") or {} m = ev.get("market") or {} live = v.get("live") liveness = "LIVE" if live is True else "REPLAY" if live is False else ( "INCONCLUSIVE" if v.get("replay_checked") else "NOT CHECKED") mv = m.get("signed_return_30m_pct") return (f"{ev.get('ts_utc')} {str(ev.get('ticker') or '?'):<6} " f"{str(ev.get('direction') or '-'):<6} " f"src={ev.get('source', 'live')} liveness={liveness} " f"sources={v.get('sources')} " f"30m={'-' if mv is None else f'{mv:+.3f}%'}\n" f" “{ev.get('quote') or ''}”") def tradeable(ev: dict) -> tuple[bool, str]: """Would this event be eligible for execution at all? Two hard refusals that have nothing to do with your risk appetite: a reconstruction is not a live observation, and an unproven liveness verdict is the single highest-frequency way to be fooled by a rerun. """ if ev.get("source") == "replay": return False, "reconstructed from archived audio, never executed" v = ev.get("verification") or {} if v.get("live") is not True: return False, "liveness not confirmed LIVE" if not ev.get("direction"): return False, "no classified direction" if not ev.get("ticker"): return False, "no instrument" return True, "eligible" # ================================================== optional broker adapter class Broker: """Optional execution. Paper unless two explicit switches are thrown. This mirrors the publisher's own broker discipline on purpose. Limit orders only: there is no market-order code path in this class, so no flag, config mistake or future edit-by-accident can produce one. """ def __init__(self, dry_run: bool = True): self.dry_run = dry_run self.key = os.environ.get("SYGNL47_APCA_KEY", "") self.secret = os.environ.get("SYGNL47_APCA_SECRET", "") self.mode = "disabled" self.reason = "no broker credentials in environment" self.fires_today = 0 self.notional_today = 0.0 self.day = datetime.now(timezone.utc).strftime("%Y-%m-%d") if not (self.key and self.secret): return want_live = os.environ.get("SYGNL47_BROKER_MODE", "paper").lower() == "live" if not want_live: self.mode, self.reason = "paper", "paper endpoint" return # Live needs a second, deliberate switch. One stray environment # variable must not be enough to move from simulated fills to real # money. if os.environ.get("SYGNL47_LIVE_CONFIRM") != "I_UNDERSTAND_THIS_TRADES_REAL_MONEY": self.mode = "paper" self.reason = ("live requested but SYGNL47_LIVE_CONFIRM is not set, " "staying on paper") return self.mode, self.reason = "live", "LIVE endpoint, real money" @property def base(self) -> str: return LIVE_URL if self.mode == "live" else PAPER_URL @property def enabled(self) -> bool: return self.mode in ("paper", "live") and bool(self.key and self.secret) def banner(self) -> str: return (f"broker {self.mode} ({self.reason}); limit orders only; " f"${PER_FIRE_USD:.0f} per fire, ${DAILY_NOTIONAL_USD:.0f} per day, " f"{MAX_FIRES_PER_DAY} fires per day max" + ("; DRY RUN, no order will be sent" if self.dry_run else "")) def _roll_day(self) -> None: today = datetime.now(timezone.utc).strftime("%Y-%m-%d") if today != self.day: self.day, self.fires_today, self.notional_today = today, 0, 0.0 def _req(self, method: str, url: str, body: dict | None = None) -> tuple[int, dict]: req = urllib.request.Request( url, data=json.dumps(body).encode() if body else None, headers={"APCA-API-KEY-ID": self.key, "APCA-API-SECRET-KEY": self.secret, "Content-Type": "application/json"}, method=method) try: with urllib.request.urlopen(req, timeout=10) as r: return r.status, json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read() or b"{}") except (ValueError, OSError): return e.code, {} except Exception as e: # noqa: BLE001 return 0, {"error": f"{type(e).__name__}: {e}"} def quote(self, symbol: str) -> tuple[float, float, str]: code, body = self._req( "GET", f"{DATA_URL}/v2/stocks/{symbol}/quotes/latest") q = (body or {}).get("quote") or {} bid, ask = float(q.get("bp") or 0), float(q.get("ap") or 0) if bid > 0 and ask > 0: return bid, ask, "quote" return 0.0, 0.0, f"no quote (http {code})" def fire(self, symbol: str, side: str) -> str: """Place ONE bounded limit order. Returns a human-readable result.""" self._roll_day() if self.fires_today >= MAX_FIRES_PER_DAY: return f"REFUSED: daily fire cap reached ({MAX_FIRES_PER_DAY})" room = DAILY_NOTIONAL_USD - self.notional_today if room <= 0: return f"REFUSED: daily notional cap reached (${DAILY_NOTIONAL_USD:.0f})" if side not in ("buy", "sell"): return f"REFUSED: unknown side {side!r}" if not self.enabled and not self.dry_run: return f"REFUSED: {self.reason}" notional = min(PER_FIRE_USD, room) # --dry-run means NO broker contact at all, not just no order. A quote # request still carries your credentials to the broker, so it is skipped # too; the promise in --help has to be literally true. bid, ask, why = (0.0, 0.0, "dry run, no quote fetched") if self.enabled and not self.dry_run: bid, ask, why = self.quote(symbol) if bid <= 0 or ask <= 0: if not self.dry_run: return f"REFUSED: {why}" bid = ask = 0.0 off = LIMIT_OFFSET_BPS / 10_000.0 if ask > 0: limit_price = round(ask * (1 + off) if side == "buy" else bid * (1 - off), 2) qty = int(notional // limit_price) else: limit_price, qty = 0.0, 0 if self.dry_run: self.fires_today += 1 self.notional_today += notional px = f"{limit_price:.2f}" if limit_price else "unpriced" return (f"DRY RUN: would send LIMIT {side} {symbol} " f"qty {qty or '?'} at {px}, notional cap ${notional:.0f}. " f"No order was sent and no broker was contacted.") if qty < 1: return (f"REFUSED: ${notional:.0f} is less than one share at " f"{limit_price:.2f}") order = {"symbol": symbol, "qty": qty, "side": side, "type": "limit", # never market "limit_price": limit_price, "time_in_force": "day", "client_order_id": f"sygnl47-starter-{int(time.time())}"} code, body = self._req("POST", f"{self.base}/v2/orders", order) if code not in (200, 201): return f"ORDER REJECTED (http {code}): {str(body)[:160]}" self.fires_today += 1 self.notional_today += qty * limit_price return (f"{self.mode.upper()} LIMIT {side} {symbol} qty {qty} at " f"{limit_price:.2f}, id {body.get('id', '?')}") # ==================================================================== loop def run(args: argparse.Namespace) -> int: feed = Feed(args.base) broker = Broker(dry_run=args.dry_run) if args.execute else None print("=" * 78) print("SYGNL/47 starter agent") print(DISCLAIMER) print("=" * 78) log(f"base {feed.base} tier {args.tier} " f"(${TIER_PRICE_USD[args.tier]:.2f} per call) every {args.interval}s") if feed.sub_pass: log(f"payment: subscription pass in {PASS_HEADER}") elif feed.payment: log(f"payment: per-call payload in {PAY_HEADER}") else: log("payment: none configured. Priced calls will report their 402 " "requirements and nothing will be charged.") if broker is not None: log(broker.banner()) if broker.mode == "live": log("LIVE MODE. Real money. Nothing about this signal is evidenced.") else: log("execution: off. Observation only. Pass --execute to enable it.") if args.manifest: m = feed.manifest() inv = (m or {}).get("inventory") or {} log("free manifest:") print(json.dumps({ "not_sold": m.get("not_sold"), "inventory": inv, "tiers": [{"tier": t.get("tier"), "route": t.get("route"), "price_usd": t.get("price_usd")} for t in ((m.get("pricing_model") or {}).get("tiers") or [])], }, indent=2)) return 0 seen: set[str] = set() first_pass = True while True: state, body = feed.poll(args.tier, args.limit) if state == "ok": events = body.get("events") or [] sync = body.get("sync") or {} fresh = sync.get("feed_age_seconds") new = [e for e in events if e.get("event_id") not in seen] for e in events: if e.get("event_id"): seen.add(e["event_id"]) log(f"{len(events)} event(s) visible at tier {args.tier}, " f"{len(new)} new, publisher artifact {fresh}s old, " f"{body.get('withheld_newer_than_tier', 0)} withheld as newer " f"than this tier") if first_pass and new: log("first pass: the events below are backlog, not fresh fires") for e in new: print(event_line(e)) ok, why = tradeable(e) if broker is None: continue if not ok: log(f" no execution: {why}") continue if first_pass and not args.execute_backlog: log(" no execution: backlog on first pass. " "Pass --execute-backlog if you really want this.") continue side = "buy" if e.get("direction") == "long" else "sell" log(" " + broker.fire(str(e["ticker"]), side)) first_pass = False elif state == "empty": log("nothing inside this tier's window. HTTP 404, not settled, " "you were not charged.") first_pass = False elif state == "unavailable": log("publisher sync unavailable. HTTP 503, not settled, you were " "not charged. The seller refuses to serve a stale artifact as " "though it were current.") elif state == "payment": log(describe_402(body)) if body.get("token_error"): log(f"pass rejected: {body['token_error']}") else: log(f"unexpected response: {json.dumps(body)[:200]}") if args.once: return 0 time.sleep(max(5, args.interval)) def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser( prog="agent.py", description=("SYGNL/47 starter agent. Polls a priced tier of the " "presidential statement observation feed, handles the " "x402 402 and retry, prints new events. Optional broker " "execution is off by default and paper-only until you " "throw two explicit switches."), epilog=("This feed has ZERO confirmed live outcomes. n=2, and both " "events are reconstructed from archived broadcast audio. You " "are buying an observation feed, not a strategy. Nothing here " "is evidenced, and any real money you route through it is " "entirely your own risk."), formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("--base", default=BASE, help="service base URL") p.add_argument("--tier", choices=sorted(TIER_ROUTES), default="archive", help="which rung of the ladder to poll (default: archive, " "the cheapest at $0.10 per call)") p.add_argument("--interval", type=float, default=60.0, help="seconds between polls, minimum 5 (default: 60)") p.add_argument("--limit", type=int, default=100, help="max events per response, 1 to 500 (default: 100)") p.add_argument("--once", action="store_true", help="poll once and exit") p.add_argument("--manifest", action="store_true", help="print the free service manifest and inventory, then exit") p.add_argument("--execute", action="store_true", help="enable the optional broker path. Paper unless " "SYGNL47_BROKER_MODE=live AND SYGNL47_LIVE_CONFIRM are " "both set. Limit orders only, always.") p.add_argument("--execute-backlog", action="store_true", help="also act on events already on file at startup. Off by " "default, because a backlog is history, not a fire.") p.add_argument("--dry-run", action="store_true", help="never contact a broker and never send an order. " "Prints the order that would have been built.") args = p.parse_args(argv) if args.execute and not args.dry_run and os.environ.get( "SYGNL47_BROKER_MODE", "paper").lower() == "live": log("live execution requested") try: return run(args) except KeyboardInterrupt: print() log("stopped") return 0 if __name__ == "__main__": sys.exit(main())