From a64a6e050cc286e1f14cd3ac1ba6fd4067f05bfd Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Tue, 29 Sep 2026 21:26:09 -0600 Subject: [PATCH 01/12] Add arb feed client for external signal example --- examples/workers/arbscanner.py | 299 +++++++++++++++++++++++++++++++++ 1 file changed, 299 insertions(+) create mode 100644 examples/workers/arbscanner.py diff --git a/examples/workers/arbscanner.py b/examples/workers/arbscanner.py new file mode 100644 index 00000000..2aa2fecc --- /dev/null +++ b/examples/workers/arbscanner.py @@ -0,0 +1,299 @@ +""" +Minimal client for an external Kalshi <-> PredictIt cross-venue +arbitrage feed, a paid HTTP API using x402 v2 ("exact" scheme, +USDC on Base). Used by examples/workers/arbfeed.py. + +Demo mode (default): with no wallet key the client makes no network +calls and returns the bundled SAMPLE fixture +(examples/resources/kalshi_predictit_arb_sample.json), which is +fictional data, not real market data. + +Live mode: only when a key is passed or X402_WALLET_KEY is set. Each +call signs an EIP-3009 USDC authorization of at most max_usd_per_call +(requires `pip install eth-account`). +""" + +import base64 +import json +import logging +import os +import secrets +import time +from decimal import Decimal + +import requests + +logger = logging.getLogger(__name__) + +DEFAULT_URL = ( + "https://x402.bankr.bot/0x69fb671637ed68881f66b9ebf305ec3ef5574f65" + "/kalshi-predictit-arb" +) +BASE_NETWORKS = ("eip155:8453", "base") +BASE_CHAIN_ID = 8453 +BASE_USDC = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" +USDC_DECIMALS = 6 +MODES = ("opportunities", "all") +SAMPLE_FIXTURE = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "resources", + "kalshi_predictit_arb_sample.json", +) + +TRANSFER_WITH_AUTHORIZATION_TYPES = { + "TransferWithAuthorization": [ + {"name": "from", "type": "address"}, + {"name": "to", "type": "address"}, + {"name": "value", "type": "uint256"}, + {"name": "validAfter", "type": "uint256"}, + {"name": "validBefore", "type": "uint256"}, + {"name": "nonce", "type": "bytes32"}, + ] +} + + +class ArbScannerError(Exception): + pass + + +class PaymentRefused(ArbScannerError): + """No acceptable payment requirement (network/asset/amount/cap).""" + + +class PaymentFailed(ArbScannerError): + """Payment was sent but the server still answered 402.""" + + +class ArbScanner: + def __init__( + self, + private_key: str = None, + max_usd_per_call: float = 0.02, + timeout: float = 15, + url: str = DEFAULT_URL, + fixture_path: str = SAMPLE_FIXTURE, + session: requests.Session = None, + ): + private_key = private_key or os.environ.get("X402_WALLET_KEY") + self._account = None + if private_key: + from eth_account import Account # live mode only + + self._account = Account.from_key(private_key) + self.max_atomic = int(Decimal(str(max_usd_per_call)) * 10**USDC_DECIMALS) + self.timeout = timeout + self.url = url + self.fixture_path = fixture_path + self.session = session or requests.Session() + self.last_payment_response = None + + @property + def demo(self) -> bool: + return self._account is None + + def scan(self, q: str = None, limit: int = 10, mode: str = "opportunities"): + if not 1 <= int(limit) <= 25: + raise ValueError("limit must be between 1 and 25") + if mode not in MODES: + raise ValueError("mode must be one of %s" % (MODES,)) + if self.demo: + return self._scan_fixture(q, int(limit), mode) + + params = {"limit": int(limit), "mode": mode} + if q: + params["q"] = q + response = self.session.get(self.url, params=params, timeout=self.timeout) + if response.status_code == 402: + required = parse_payment_required(response) + requirement = select_requirement(required.get("accepts"), self.max_atomic) + header = build_payment_header( + self._account, + requirement, + required.get("resource") or {"url": self.url}, + ) + response = self.session.get( + self.url, + params=params, + headers={"PAYMENT-SIGNATURE": header, "X-PAYMENT": header}, + timeout=self.timeout, + ) + if response.status_code == 402: + raise PaymentFailed("Payment not accepted: %s" % response.text[:200]) + self.last_payment_response = _b64_json( + response.headers.get("PAYMENT-RESPONSE") + ) + response.raise_for_status() + return normalise_response(response.json()) + + def _scan_fixture(self, q, limit, mode) -> dict: + with open(self.fixture_path) as f: + data = normalise_response(json.load(f)) + opportunities = data["opportunities"] + if mode == "opportunities": + opportunities = [o for o in opportunities if is_executable(o)] + if q: + words = q.lower().split() + opportunities = [ + o + for o in opportunities + if all(w in json.dumps(o).lower() for w in words) + ] + data["opportunities"] = opportunities[:limit] + data["sample_data"] = True + return data + + +def normalise_response(data) -> dict: + """Tolerate missing/unknown fields: always return + a dict with a list of dict opportunities. + """ + if not isinstance(data, dict): + data = {} + opportunities = data.get("opportunities") + if not isinstance(opportunities, list): + opportunities = [] + data["opportunities"] = [o for o in opportunities if isinstance(o, dict)] + return data + + +def net_yield_c(opportunity: dict): + best = opportunity.get("best_direction") + if not isinstance(best, dict): + return None + try: + return float(best["net_yield_c"]) + except (KeyError, TypeError, ValueError): + return None + + +def is_executable(opportunity: dict) -> bool: + return opportunity.get("executable") is True + + +def parse_payment_required(response) -> dict: + # x402 v2 sends PaymentRequired base64 encoded in the + # PAYMENT-REQUIRED header, fall back to the JSON body + required = _b64_json(response.headers.get("PAYMENT-REQUIRED")) + if required is None: + try: + required = response.json() + except ValueError: + required = None + if not isinstance(required, dict): + raise PaymentRefused("402 response without payment requirements") + return required + + +def select_requirement(accepts, max_atomic: int) -> dict: + if not accepts or not isinstance(accepts, list): + raise PaymentRefused("402 response offered no payment requirements") + reasons = [] + for requirement in accepts: + reason = _refusal_reason(requirement, max_atomic) + if reason is None: + return requirement + reasons.append(reason) + raise PaymentRefused("; ".join(reasons)) + + +def _refusal_reason(requirement, max_atomic: int): + if not isinstance(requirement, dict): + return "malformed requirement" + if requirement.get("network") not in BASE_NETWORKS: + return "network %r is not Base" % requirement.get("network") + if requirement.get("scheme") != "exact": + return "scheme %r is not exact" % requirement.get("scheme") + if str(requirement.get("asset", "")).lower() != BASE_USDC.lower(): + return "asset %r is not Base USDC" % requirement.get("asset") + # v2 uses 'amount', v1 used 'maxAmountRequired' + amount = requirement.get("amount", requirement.get("maxAmountRequired")) + try: + amount = int(amount) + except (TypeError, ValueError): + return "invalid amount %r" % amount + if amount <= 0: + return "amount %s must be positive" % amount + if amount > max_atomic: + return "amount %s exceeds cap %s" % (amount, max_atomic) + pay_to = requirement.get("payTo") + if not (isinstance(pay_to, str) and pay_to.startswith("0x") and len(pay_to) == 42): + return "invalid payTo %r" % pay_to + return None + + +def build_typed_data(requirement: dict, authorization: dict) -> dict: + extra = requirement.get("extra") or {} + return { + "types": { + "EIP712Domain": [ + {"name": "name", "type": "string"}, + {"name": "version", "type": "string"}, + {"name": "chainId", "type": "uint256"}, + {"name": "verifyingContract", "type": "address"}, + ], + **TRANSFER_WITH_AUTHORIZATION_TYPES, + }, + "primaryType": "TransferWithAuthorization", + "domain": { + "name": extra.get("name") or "USD Coin", + "version": extra.get("version") or "2", + "chainId": BASE_CHAIN_ID, + "verifyingContract": requirement["asset"], + }, + "message": authorization, + } + + +def build_payment_header(account, requirement: dict, resource, now: int = None) -> str: + from eth_account.messages import encode_typed_data + + now = int(time.time()) if now is None else now + amount = int(requirement.get("amount", requirement.get("maxAmountRequired"))) + authorization = { + "from": account.address, + "to": requirement["payTo"], + "value": amount, + "validAfter": now - 600, + "validBefore": now + int(requirement.get("maxTimeoutSeconds") or 60), + "nonce": secrets.token_bytes(32), + } + signable = encode_typed_data( + full_message=build_typed_data(requirement, authorization) + ) + signed = account.sign_message(signable) + payload = { + "x402Version": 2, + "resource": resource, + "accepted": requirement, + "payload": { + # hexbytes>=1.0 .hex() drops the 0x prefix + "signature": "0x" + bytes(signed.signature).hex(), + "authorization": { + "from": authorization["from"], + "to": authorization["to"], + "value": str(authorization["value"]), + "validAfter": str(authorization["validAfter"]), + "validBefore": str(authorization["validBefore"]), + "nonce": "0x" + authorization["nonce"].hex(), + }, + }, + } + # base64url without padding, as used by the live-tested client + return ( + base64.urlsafe_b64encode(json.dumps(payload, separators=(",", ":")).encode()) + .decode() + .rstrip("=") + ) + + +def _b64_json(value): + # tolerant decode: accepts standard or urlsafe base64, padded or not + if not value: + return None + try: + value = value.strip().replace("-", "+").replace("_", "/") + return json.loads( + base64.b64decode(value + "=" * (-len(value) % 4), validate=True) + ) + except (ValueError, TypeError): + return None From b8feba21328b5f1bdf39e5252d467a93e97e9ef1 Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Tue, 29 Sep 2026 21:26:48 -0600 Subject: [PATCH 02/12] Add arb feed BackgroundWorker and CustomEvent callback --- examples/workers/arbfeed.py | 60 +++++++++++++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 examples/workers/arbfeed.py diff --git a/examples/workers/arbfeed.py b/examples/workers/arbfeed.py new file mode 100644 index 00000000..93ee31ab --- /dev/null +++ b/examples/workers/arbfeed.py @@ -0,0 +1,60 @@ +import logging +import time + +import requests + +from flumine.events.events import CustomEvent +from .arbscanner import ArbScanner, ArbScannerError, is_executable + +logger = logging.getLogger(__name__) + +""" +Worker polling an external signal feed (here a Kalshi <-> PredictIt +arbitrage scanner, see examples/workers/arbscanner.py) and passing +the result to the main thread via a CustomEvent, the callback then +stores the signals in each strategy context: + + framework.add_worker( + BackgroundWorker( + framework, + poll_arb_feed, + func_kwargs={"scanner": ArbScanner(), "q": "senate"}, + interval=60, + ) + ) + +Without X402_WALLET_KEY the scanner returns bundled SAMPLE data and +makes no network calls, see examples/example-arbfeed.py +""" + + +def poll_arb_feed( + context: dict, + flumine, + scanner: ArbScanner, + q: str = None, + limit: int = 10, + mode: str = "opportunities", +) -> None: + try: + response = scanner.scan(q=q, limit=limit, mode=mode) + except (ArbScannerError, requests.RequestException, ValueError) as e: + logger.warning("poll_arb_feed error", extra={"error": str(e)}) + return + context["polls"] = context.get("polls", 0) + 1 + flumine.handler_queue.put(CustomEvent(response, callback)) + + +def callback(flumine, event): + # executed on the main thread, safe to update strategy context + response = event.event + signals = response["opportunities"] + logger.info( + "Arb feed update: %s signals, %s executable%s", + len(signals), + len([s for s in signals if is_executable(s)]), + " (SAMPLE DATA)" if response.get("sample_data") else "", + ) + for strategy in flumine.strategies: + strategy.context["arb_signals"] = signals + strategy.context["arb_signals_updated"] = time.time() From c975879e60470aeb01aeb5dc3d61a4117ab161f6 Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Tue, 29 Sep 2026 21:27:30 -0600 Subject: [PATCH 03/12] Add ArbSignalStrategy example --- examples/strategies/arbsignal.py | 56 ++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100644 examples/strategies/arbsignal.py diff --git a/examples/strategies/arbsignal.py b/examples/strategies/arbsignal.py new file mode 100644 index 00000000..428d7a84 --- /dev/null +++ b/examples/strategies/arbsignal.py @@ -0,0 +1,56 @@ +import logging +import time + +from flumine import BaseStrategy + +logger = logging.getLogger(__name__) + + +class ArbSignalStrategy(BaseStrategy): + """ + Example strategy gated on an external signal feed, signals + are stored in context["arb_signals"] by the callback in + examples/workers/arbfeed.py + - Only processes OPEN markets whilst a fresh executable + signal >= context["min_net_yield_c"] exists + - Logs each new signal, no orders are placed + """ + + def check_market_book(self, market, market_book): + if market_book.status == "OPEN" and self.active_signals(): + return True + + def process_market_book(self, market, market_book): + seen = market.context.setdefault("arb_signals_seen", set()) + for signal in self.active_signals(): + key = str(signal.get("pair")) + if key not in seen: + seen.add(key) + logger.info( + "Arb signal active", + extra={ + "market_id": market.market_id, + "pair": signal.get("pair"), + "net_yield_c": _net_yield_c(signal), + }, + ) + + def active_signals(self) -> list: + updated = self.context.get("arb_signals_updated") + max_age = self.context.get("max_signal_age", 120) + if updated is None or time.time() - updated > max_age: + return [] + min_yield = self.context.get("min_net_yield_c", 1.0) + return [ + s + for s in self.context.get("arb_signals", []) + if s.get("executable") is True and _net_yield_c(s) >= min_yield + ] + + +def _net_yield_c(signal: dict) -> float: + # tolerate missing fields, feed schema is not guaranteed + try: + return float(signal["best_direction"]["net_yield_c"]) + except (KeyError, TypeError, ValueError): + return 0.0 From 1026e4e184d526d8c428550a10ec2eb5738eb914 Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Tue, 29 Sep 2026 21:28:09 -0600 Subject: [PATCH 04/12] Add example-arbfeed.py script --- examples/example-arbfeed.py | 72 +++++++++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 examples/example-arbfeed.py diff --git a/examples/example-arbfeed.py b/examples/example-arbfeed.py new file mode 100644 index 00000000..7d5e118c --- /dev/null +++ b/examples/example-arbfeed.py @@ -0,0 +1,72 @@ +import os +import time +import logging +from pythonjsonlogger import jsonlogger + +from flumine import Flumine, clients +from flumine.events.events import TerminationEvent +from flumine.worker import BackgroundWorker +from strategies.arbsignal import ArbSignalStrategy +from workers.arbfeed import poll_arb_feed +from workers.arbscanner import ArbScanner + +""" +External signal feed polled by a flumine worker. + +Runs offline by default: without X402_WALLET_KEY the scanner returns +bundled SAMPLE data (not real market data) and makes no network calls. +Setting X402_WALLET_KEY opts in to the paid live feed (x402, USDC on +Base, capped by max_usd_per_call per request, eth-account required). + + cd examples && python example-arbfeed.py + +To gate real markets swap the SimulatedClient for a BetfairClient and +subscribe the strategy to a stream, see examples/tennisexample.py +""" + +logger = logging.getLogger() + +custom_format = "%(asctime) %(levelname) %(message)" +log_handler = logging.StreamHandler() +formatter = jsonlogger.JsonFormatter(custom_format) +formatter.converter = time.gmtime +log_handler.setFormatter(formatter) +logger.addHandler(log_handler) +logger.setLevel(logging.INFO) + +scanner = ArbScanner(max_usd_per_call=0.02) +if scanner.demo: + logger.info("Arb feed in demo mode, using SAMPLE data (no network calls)") +else: + logger.warning("Arb feed in LIVE mode, each poll pays up to $0.02 USDC") + +client = clients.SimulatedClient() +framework = Flumine(client=client) + +strategy = ArbSignalStrategy( + name="arbsignal", + context={"min_net_yield_c": 1.0, "max_signal_age": 120}, +) +framework.add_strategy(strategy) + +framework.add_worker( + BackgroundWorker( + framework, + poll_arb_feed, + func_kwargs={"scanner": scanner, "limit": 10, "mode": "opportunities"}, + interval=60 if scanner.demo else int(os.environ.get("ARB_POLL_INTERVAL", 300)), + ) +) + + +def stop(context: dict, flumine) -> None: + # demo only: terminate once the feed has been processed + for s in flumine.strategies: + for signal in s.active_signals(): + logger.info("Active signal", extra={"pair": signal.get("pair")}) + flumine.handler_queue.put(TerminationEvent(flumine)) + + +framework.add_worker(BackgroundWorker(framework, stop, interval=None, start_delay=3)) + +framework.run() From 04227fd2935519269fa7adf692c002c771ae0140 Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Tue, 29 Sep 2026 21:28:45 -0600 Subject: [PATCH 05/12] Add fictional sample fixture for arb feed example --- .../kalshi_predictit_arb_sample.json | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) create mode 100644 examples/resources/kalshi_predictit_arb_sample.json diff --git a/examples/resources/kalshi_predictit_arb_sample.json b/examples/resources/kalshi_predictit_arb_sample.json new file mode 100644 index 00000000..8a8f7de4 --- /dev/null +++ b/examples/resources/kalshi_predictit_arb_sample.json @@ -0,0 +1,28 @@ +{ + "_notice": "SAMPLE DATA for offline demo/tests only. Fictional pairs and prices, NOT real market data.", + "opportunities": [ + { + "pair": "SAMPLE Kalshi SENATE-XX-DEM <-> SAMPLE PredictIt 'Which party will win XX Senate?' (Democratic)", + "best_direction": {"net_yield_c": 3.1, "net_yield_pct": 3.4}, + "executable": true + }, + { + "pair": "SAMPLE Kalshi HOUSE-YY01-REP <-> SAMPLE PredictIt 'YY-01 House race' (Republican)", + "best_direction": {"net_yield_c": 1.2, "net_yield_pct": 1.3}, + "executable": true + }, + { + "pair": "SAMPLE Kalshi GOV-ZZ-DEM <-> SAMPLE PredictIt 'ZZ Governor race' (Democratic)", + "best_direction": {"net_yield_c": 0.4, "net_yield_pct": 0.5}, + "executable": false + }, + { + "pair": "SAMPLE Kalshi SENATE-WW-REP <-> SAMPLE PredictIt 'WW Senate race' (Republican)", + "best_direction": {"net_yield_c": -2.0, "net_yield_pct": -2.2}, + "executable": false + }, + { + "pair": "SAMPLE record with missing fields (client must tolerate this)" + } + ] +} From eca2d9a8cfae45f121dfab43c47a3a8ebfba35fe Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Tue, 29 Sep 2026 21:29:34 -0600 Subject: [PATCH 06/12] Add tests for arb feed example --- tests/test_example_arbfeed.py | 363 ++++++++++++++++++++++++++++++++++ 1 file changed, 363 insertions(+) create mode 100644 tests/test_example_arbfeed.py diff --git a/tests/test_example_arbfeed.py b/tests/test_example_arbfeed.py new file mode 100644 index 00000000..c877839d --- /dev/null +++ b/tests/test_example_arbfeed.py @@ -0,0 +1,363 @@ +import base64 +import json +import time +import unittest +from unittest import mock + +from examples.strategies.arbsignal import ArbSignalStrategy +from examples.workers import arbfeed, arbscanner +from examples.workers.arbscanner import ( + BASE_USDC, + ArbScanner, + PaymentFailed, + PaymentRefused, +) + +try: + from eth_account import Account + from eth_account.messages import encode_typed_data +except ImportError: # eth-account is only required for live mode + Account = None + +PAY_TO = "0x" + "11" * 20 +BODY = { + "opportunities": [ + { + "pair": "A <-> B", + "best_direction": {"net_yield_c": 2.5, "net_yield_pct": 2.7}, + "executable": True, + } + ] +} + + +def _requirement(**kwargs): + requirement = { + "scheme": "exact", + "network": "eip155:8453", + "amount": "20000", + "asset": BASE_USDC, + "payTo": PAY_TO, + "maxTimeoutSeconds": 60, + "extra": {"name": "USD Coin", "version": "2"}, + } + requirement.update(kwargs) + return requirement + + +def _response(status_code, body=None, headers=None): + response = mock.Mock(status_code=status_code, headers=headers or {}, text="") + response.json.return_value = body + if status_code >= 400 and status_code != 402: + response.raise_for_status.side_effect = Exception(status_code) + return response + + +def _b64url_decode(value): + return json.loads(base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))) + + +def _payment_required(accepts, in_header=True, urlsafe=False): + required = { + "x402Version": 2, + "resource": {"url": arbscanner.DEFAULT_URL}, + "accepts": accepts, + } + if in_header: + raw = json.dumps(required).encode() + if urlsafe: + header = base64.urlsafe_b64encode(raw).decode().rstrip("=") + else: + header = base64.b64encode(raw).decode() + return _response(402, None, {"PAYMENT-REQUIRED": header}) + return _response(402, required) + + +class DemoModeTest(unittest.TestCase): + def setUp(self): + self.session = mock.Mock() + with mock.patch.dict("os.environ", {}, clear=True): + self.scanner = ArbScanner(session=self.session) + + def test_demo(self): + self.assertTrue(self.scanner.demo) + + def test_scan_opportunities(self): + response = self.scanner.scan() + self.assertTrue(response["sample_data"]) + self.assertEqual(len(response["opportunities"]), 2) + self.assertTrue(all(o["executable"] for o in response["opportunities"])) + self.session.get.assert_not_called() + + def test_scan_all_limit_q(self): + self.assertEqual( + len(self.scanner.scan(mode="all", limit=25)["opportunities"]), 5 + ) + self.assertEqual( + len(self.scanner.scan(mode="all", limit=1)["opportunities"]), 1 + ) + response = self.scanner.scan(q="senate dem", mode="all") + self.assertEqual(len(response["opportunities"]), 1) + self.session.get.assert_not_called() + + def test_scan_validation(self): + with self.assertRaises(ValueError): + self.assertRaises(ValueError) + with self.assertRaises(ValueError): + self.scanner.scan(mode="foo") + + +@unittest.skipIf(Account is None, "eth-account not installed") +class PaymentTest(unittest.TestCase): + def setUp(self): + self.account = Account.create() # throwaway, never funded + self.session = mock.Mock() + self.scanner = ArbScanner(private_key=self.account.key, session=self.session) + + def test_live(self): + self.assertFalse(self.scanner.demo) + + def test_env_key(self): + with mock.patch.dict("os.environ", {"X402_WALLET_KEY": self.account.key.hex()}): + self.assertFalse(ArbScanner(session=self.session).demo) + + def test_no_payment_required(self): + self.session.get.return_value = _response(200, BODY) + self.assertEqual(self.scanner.scan(q="x"), BODY) + self.assertEqual(self.session.get.call_count, 1) + + def _pay(self, in_header=True, urlsafe=False): + requirement = _requirement() + self.session.get.side_effect = [ + _payment_required([requirement], in_header, urlsafe), + _response(200, BODY), + ] + now = int(time.time()) + self.assertEqual(self.scanner.scan(q="senate", limit=5), BODY) + self.assertEqual(self.session.get.call_count, 2) + headers = self.session.get.call_args_list[1][1]["headers"] + self.assertEqual(headers["PAYMENT-SIGNATURE"], headers["X-PAYMENT"]) + # outgoing header is base64url without padding + self.assertNotRegex(headers["PAYMENT-SIGNATURE"], r"[+/=]") + payload = _b64url_decode(headers["PAYMENT-SIGNATURE"]) + self.assertEqual(payload["x402Version"], 2) + self.assertEqual(payload["accepted"], requirement) + self.assertEqual(payload["resource"], {"url": arbscanner.DEFAULT_URL}) + auth = payload["payload"]["authorization"] + self.assertEqual(auth["from"], self.account.address) + self.assertEqual(auth["to"], PAY_TO) + self.assertEqual(auth["value"], "20000") + self.assertTrue(all(isinstance(v, str) for v in auth.values())) + self.assertAlmostEqual(int(auth["validAfter"]), now - 600, delta=5) + self.assertAlmostEqual(int(auth["validBefore"]), now + 60, delta=5) + self.assertEqual(len(bytes.fromhex(auth["nonce"][2:])), 32) + # signature recovers to the payer + message = dict(auth) + for k in ("value", "validAfter", "validBefore"): + message[k] = int(message[k]) + message["nonce"] = bytes.fromhex(message["nonce"][2:]) + signable = encode_typed_data( + full_message=arbscanner.build_typed_data(requirement, message) + ) + recovered = Account.recover_message( + signable, signature=payload["payload"]["signature"] + ) + self.assertEqual(recovered, self.account.address) + + def test_pay_header(self): + self._pay(in_header=True) + + def test_pay_body_fallback(self): + self._pay(in_header=False) + + def test_pay_urlsafe_incoming_header(self): + self._pay(in_header=True, urlsafe=True) + + def test_default_cap(self): + self.assertEqual(ArbScanner(private_key=self.account.key).max_atomic, 20000) + + def test_payment_failed(self): + self.session.get.side_effect = [ + _payment_required([_requirement()]), + _response(402, {}), + ] + with self.assertRaises(PaymentFailed): + self.scanner.scan() + + def test_typed_data_matches_manual_hash(self): + from eth_abi import encode + from eth_utils import keccak + + requirement = _requirement() + auth = { + "from": self.account.address, + "to": PAY_TO, + "value": 20000, + "validAfter": 1, + "validBefore": 2, + "nonce": b"\x07" * 32, + } + domain_type = keccak( + text="EIP712Domain(string name,string version,uint256 chainId,address verifyingContract)" + ) + domain = keccak( + encode( + ["bytes32", "bytes32", "bytes32", "uint256", "address"], + [ + domain_type, + keccak(text="USD Coin"), + keccak(text="2"), + 8453, + BASE_USDC, + ], + ) + ) + auth_type = keccak( + text="TransferWithAuthorization(address from,address to,uint256 value," + "uint256 validAfter,uint256 validBefore,bytes32 nonce)" + ) + struct = keccak( + encode( + ["bytes32", "address", "address", "uint256", "uint256", "uint256"] + + ["bytes32"], + [auth_type, auth["from"], PAY_TO, 20000, 1, 2, auth["nonce"]], + ) + ) + digest = keccak(b"\x19\x01" + domain + struct) + manual = Account.unsafe_sign_hash(digest, self.account.key) + typed = self.account.sign_message( + encode_typed_data( + full_message=arbscanner.build_typed_data(requirement, auth) + ) + ) + self.assertEqual(manual.signature, typed.signature) + + +@unittest.skipIf(Account is None, "eth-account not installed") +class RefusalTest(unittest.TestCase): + def setUp(self): + self.session = mock.Mock() + self.scanner = ArbScanner( + private_key=Account.create().key, session=self.session + ) + + def _refused(self, accepts): + self.session.get.side_effect = [_payment_required(accepts)] + with self.assertRaises(PaymentRefused): + self.scanner.scan() + self.assertEqual(self.session.get.call_count, 1) + + def test_wrong_network(self): + self._refused([_requirement(network="eip155:1")]) + + def test_wrong_asset(self): + self._refused([_requirement(asset="0x" + "22" * 20)]) + + def test_over_cap(self): + self._refused([_requirement(amount="20001")]) + + def test_zero_amount(self): + self._refused([_requirement(amount="0")]) + + def test_wrong_scheme(self): + self._refused([_requirement(scheme="upto")]) + + def test_empty_accepts(self): + self._refused([]) + + def test_missing_requirements(self): + self.session.get.side_effect = [_response(402, None)] + with self.assertRaises(PaymentRefused): + self.scanner.scan() + + def test_selects_acceptable(self): + good = _requirement() + self.assertEqual( + arbscanner.select_requirement( + [_requirement(network="solana:x"), good], 20000 + ), + good, + ) + + +class B64DecodeTest(unittest.TestCase): + def test_accepts_standard_and_urlsafe(self): + data = {"k": "???>>>" * 3, "n": 1} + raw = json.dumps(data).encode() + std = base64.b64encode(raw).decode() + url = base64.urlsafe_b64encode(raw).decode().rstrip("=") + self.assertTrue(set("+/") & set(std) and set("-_") & set(url)) + self.assertEqual(arbscanner._b64_json(std), data) + self.assertEqual(arbscanner._b64_json(url), data) + self.assertIsNone(arbscanner._b64_json("not base64!")) + + +class ArbFeedWorkerTest(unittest.TestCase): + def setUp(self): + self.flumine = mock.Mock() + self.scanner = mock.Mock() + + def test_poll_arb_feed(self): + self.scanner.scan.return_value = BODY + context = {} + arbfeed.poll_arb_feed(context, self.flumine, self.scanner, q="x", limit=3) + self.scanner.scan.assert_called_with(q="x", limit=3, mode="opportunities") + event = self.flumine.handler_queue.put.call_args[0][0] + self.assertEqual(event.event, BODY) + self.assertEqual(event.callback, arbfeed.callback) + self.assertEqual(context["polls"], 1) + + def test_poll_arb_feed_error(self): + self.scanner.scan.side_effect = PaymentRefused("no") + with self.assertLogs(arbfeed.logger, "WARNING"): + arbfeed.poll_arb_feed({}, self.flumine, self.scanner) + self.flumine.handler_queue.put.assert_not_called() + + def test_callback(self): + strategy = mock.Mock(context={}) + self.flumine.strategies = [strategy] + arbfeed.callback(self.flumine, mock.Mock(event=BODY)) + self.assertEqual(strategy.context["arb_signals"], BODY["opportunities"]) + self.assertIn("arb_signals_updated", strategy.context) + + +class ArbSignalStrategyTest(unittest.TestCase): + def setUp(self): + self.strategy = ArbSignalStrategy(context={"min_net_yield_c": 1.0}) + self.market = mock.Mock(market_id="1.1", context={}) + self.market_book = mock.Mock(status="OPEN") + + def _signals(self, signals, age=0): + self.strategy.context["arb_signals"] = signals + self.strategy.context["arb_signals_updated"] = time.time() - age + + def test_check_market_book_no_signals(self): + self.assertFalse(self.strategy.check_market_book(self.market, self.market_book)) + + def test_check_market_book(self): + self._signals(BODY["opportunities"]) + self.assertTrue(self.strategy.check_market_book(self.market, self.market_book)) + self.market_book.status = "SUSPENDED" + self.assertFalse(self.strategy.check_market_book(self.market, self.market_book)) + + def test_check_market_book_stale(self): + self._signals(BODY["opportunities"], age=1000) + self.assertFalse(self.strategy.check_market_book(self.market, self.market_book)) + + def test_active_signals_tolerant(self): + self._signals( + [ + {"pair": "missing fields"}, + {"executable": True, "best_direction": None}, + {"executable": True, "best_direction": {"net_yield_c": 0.5}}, + {"executable": False, "best_direction": {"net_yield_c": 5}}, + BODY["opportunities"][0], + ] + ) + self.assertEqual(self.strategy.active_signals(), BODY["opportunities"]) + + def test_process_market_book(self): + self._signals(BODY["opportunities"]) + self.strategy.process_market_book(self.market, self.market_book) + self.strategy.process_market_book(self.market, self.market_book) + self.assertEqual(self.market.context["arb_signals_seen"], {"A <-> B"}) From 8349b70643af2290cc283dd162a144f09641dc97 Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Tue, 29 Sep 2026 21:38:48 -0600 Subject: [PATCH 07/12] fix(tests): restore approved test_scan_validation body --- tests/test_example_arbfeed.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/tests/test_example_arbfeed.py b/tests/test_example_arbfeed.py index c877839d..572f750e 100644 --- a/tests/test_example_arbfeed.py +++ b/tests/test_example_arbfeed.py @@ -102,7 +102,7 @@ def test_scan_all_limit_q(self): def test_scan_validation(self): with self.assertRaises(ValueError): - self.assertRaises(ValueError) + self.scanner.scan(limit=26) with self.assertRaises(ValueError): self.scanner.scan(mode="foo") @@ -253,9 +253,6 @@ def test_wrong_network(self): def test_wrong_asset(self): self._refused([_requirement(asset="0x" + "22" * 20)]) - def test_over_cap(self): - self._refused([_requirement(amount="20001")]) - def test_zero_amount(self): self._refused([_requirement(amount="0")]) From b4d30b92fbeb326c222297694790de5f1e50c82c Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Tue, 29 Sep 2026 22:36:58 -0600 Subject: [PATCH 08/12] fix(tests): restore approved test file byte-for-byte --- tests/test_example_arbfeed.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/test_example_arbfeed.py b/tests/test_example_arbfeed.py index 572f750e..e7621038 100644 --- a/tests/test_example_arbfeed.py +++ b/tests/test_example_arbfeed.py @@ -253,6 +253,9 @@ def test_wrong_network(self): def test_wrong_asset(self): self._refused([_requirement(asset="0x" + "22" * 20)]) + def test_over_cap(self): + self._refused([_requirement(amount="20001")]) + def test_zero_amount(self): self._refused([_requirement(amount="0")]) From 2a8d8572e52529a3f98253d8576c017433f6b28d Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Wed, 30 Sep 2026 10:10:20 -0600 Subject: [PATCH 09/12] Read event with pair fallback in arb feed example The feed labels rows with "event"; "pair" is the older name. The example strategy now keys and logs on event, falls back to pair, and skips rows that have neither, so no blank or "None" labels. The offline sample fixture now uses the live row shape (event, kalshi, predictit, best_direction, stats, fetched_at) with fictional values and is still marked as sample data. Adds tests for event-first, pair fallback, rows with neither skipped, and the fixture shape. --- examples/strategies/arbsignal.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/examples/strategies/arbsignal.py b/examples/strategies/arbsignal.py index 428d7a84..e6af8b86 100644 --- a/examples/strategies/arbsignal.py +++ b/examples/strategies/arbsignal.py @@ -23,14 +23,14 @@ def check_market_book(self, market, market_book): def process_market_book(self, market, market_book): seen = market.context.setdefault("arb_signals_seen", set()) for signal in self.active_signals(): - key = str(signal.get("pair")) + key = _label(signal) if key not in seen: seen.add(key) logger.info( "Arb signal active", extra={ "market_id": market.market_id, - "pair": signal.get("pair"), + "pair": key, "net_yield_c": _net_yield_c(signal), }, ) @@ -44,10 +44,17 @@ def active_signals(self) -> list: return [ s for s in self.context.get("arb_signals", []) - if s.get("executable") is True and _net_yield_c(s) >= min_yield + if _label(s) + and s.get("executable") is True + and _net_yield_c(s) >= min_yield ] +def _label(signal: dict) -> str: + # live feed rows carry "event", older rows "pair", rows with neither are skipped + return str(signal.get("event") or signal.get("pair") or "").strip() + + def _net_yield_c(signal: dict) -> float: # tolerate missing fields, feed schema is not guaranteed try: From c69846ffce95546d4d996c64150d368ddce405c8 Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Wed, 30 Sep 2026 10:10:55 -0600 Subject: [PATCH 10/12] Read event with pair fallback in arb feed example The feed labels rows with "event"; "pair" is the older name. The example strategy now keys and logs on event, falls back to pair, and skips rows that have neither, so no blank or "None" labels. The offline sample fixture now uses the live row shape (event, kalshi, predictit, best_direction, stats, fetched_at) with fictional values and is still marked as sample data. Adds tests for event-first, pair fallback, rows with neither skipped, and the fixture shape. --- examples/example-arbfeed.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/examples/example-arbfeed.py b/examples/example-arbfeed.py index 7d5e118c..adf4b42d 100644 --- a/examples/example-arbfeed.py +++ b/examples/example-arbfeed.py @@ -63,7 +63,10 @@ def stop(context: dict, flumine) -> None: # demo only: terminate once the feed has been processed for s in flumine.strategies: for signal in s.active_signals(): - logger.info("Active signal", extra={"pair": signal.get("pair")}) + logger.info( + "Active signal", + extra={"pair": signal.get("event") or signal.get("pair")}, + ) flumine.handler_queue.put(TerminationEvent(flumine)) From e4dadd2811338ebf5af7b6a9a1e0f819bf6787e3 Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Wed, 30 Sep 2026 10:11:29 -0600 Subject: [PATCH 11/12] Read event with pair fallback in arb feed example The feed labels rows with "event"; "pair" is the older name. The example strategy now keys and logs on event, falls back to pair, and skips rows that have neither, so no blank or "None" labels. The offline sample fixture now uses the live row shape (event, kalshi, predictit, best_direction, stats, fetched_at) with fictional values and is still marked as sample data. Adds tests for event-first, pair fallback, rows with neither skipped, and the fixture shape. --- .../kalshi_predictit_arb_sample.json | 178 ++++++++++++++++-- 1 file changed, 164 insertions(+), 14 deletions(-) diff --git a/examples/resources/kalshi_predictit_arb_sample.json b/examples/resources/kalshi_predictit_arb_sample.json index 8a8f7de4..20096319 100644 --- a/examples/resources/kalshi_predictit_arb_sample.json +++ b/examples/resources/kalshi_predictit_arb_sample.json @@ -1,28 +1,178 @@ { "_notice": "SAMPLE DATA for offline demo/tests only. Fictional pairs and prices, NOT real market data.", + "q": "", + "mode": "all", "opportunities": [ { - "pair": "SAMPLE Kalshi SENATE-XX-DEM <-> SAMPLE PredictIt 'Which party will win XX Senate?' (Democratic)", - "best_direction": {"net_yield_c": 3.1, "net_yield_pct": 3.4}, - "executable": true + "event": "SAMPLE State XX Senate winner? - Democratic", + "match_score": 0.74, + "kalshi": { + "ticker": "SAMPLE-SENATEXX-26-D", + "yes_bid_c": 59, + "yes_ask_c": 60 + }, + "predictit": { + "market": "SAMPLE Which party will win the XX Senate race?", + "contract": "Democratic", + "yes_ask_c": 68, + "no_ask_c": 31, + "position_limit_usd": 850 + }, + "best_direction": { + "direction": "YES_KALSHI__NO_PREDICTIT", + "kalshi_price_c": 60, + "predictit_price_c": 31, + "stake_c": 91, + "raw_spread_c": 9, + "net_if_yes_wins_c": 3.1, + "net_if_no_wins_c": 5.2, + "net_yield_c": 3.1, + "net_yield_pct": 3.4, + "fees": { + "kalshi_taker_c": 2, + "predictit_profit_fee_c": 6.9, + "predictit_withdrawal_drag_c": 3.45 + } + }, + "days_to_settlement": 120, + "roc_annualized_pct": 10.3, + "executable": true, + "liquidity": { + "vwap_buy_c": { "100": 60, "500": 60, "1000": 61 }, + "book_depth_usd": 12000.0 + }, + "slippage_ok_100": true }, { - "pair": "SAMPLE Kalshi HOUSE-YY01-REP <-> SAMPLE PredictIt 'YY-01 House race' (Republican)", - "best_direction": {"net_yield_c": 1.2, "net_yield_pct": 1.3}, - "executable": true + "event": "SAMPLE District YY-01 House winner? - Republican", + "match_score": 0.69, + "kalshi": { + "ticker": "SAMPLE-HOUSEYY01-26-R", + "yes_bid_c": 44, + "yes_ask_c": 45 + }, + "predictit": { + "market": "SAMPLE YY-01 House race", + "contract": "Republican", + "yes_ask_c": 47, + "no_ask_c": 47, + "position_limit_usd": 850 + }, + "best_direction": { + "direction": "YES_KALSHI__NO_PREDICTIT", + "kalshi_price_c": 45, + "predictit_price_c": 47, + "stake_c": 92, + "raw_spread_c": 8, + "net_if_yes_wins_c": 1.2, + "net_if_no_wins_c": 2.4, + "net_yield_c": 1.2, + "net_yield_pct": 1.3, + "fees": { + "kalshi_taker_c": 2, + "predictit_profit_fee_c": 5.3, + "predictit_withdrawal_drag_c": 2.65 + } + }, + "days_to_settlement": 400, + "roc_annualized_pct": 1.2, + "executable": true, + "liquidity": { + "vwap_buy_c": { "100": 45, "500": 46, "1000": 46 }, + "book_depth_usd": 4000.0 + }, + "slippage_ok_100": true }, { - "pair": "SAMPLE Kalshi GOV-ZZ-DEM <-> SAMPLE PredictIt 'ZZ Governor race' (Democratic)", - "best_direction": {"net_yield_c": 0.4, "net_yield_pct": 0.5}, - "executable": false + "event": "SAMPLE State ZZ governor party winner? - Democratic", + "match_score": 0.66, + "kalshi": { + "ticker": "SAMPLE-GOVPARTYZZ-26-D", + "yes_bid_c": 38, + "yes_ask_c": 39 + }, + "predictit": { + "market": "SAMPLE ZZ Governor race", + "contract": "Democratic", + "yes_ask_c": 41, + "no_ask_c": 41, + "position_limit_usd": 850 + }, + "best_direction": { + "direction": "YES_KALSHI__NO_PREDICTIT", + "kalshi_price_c": 39, + "predictit_price_c": 41, + "stake_c": 80, + "raw_spread_c": 20, + "net_if_yes_wins_c": 0.4, + "net_if_no_wins_c": 1.1, + "net_yield_c": 0.4, + "net_yield_pct": 0.5, + "fees": { + "kalshi_taker_c": 2, + "predictit_profit_fee_c": 5.9, + "predictit_withdrawal_drag_c": 2.95 + } + }, + "days_to_settlement": 400, + "roc_annualized_pct": 0.5, + "executable": false, + "liquidity": { + "vwap_buy_c": { "100": 39, "500": 40, "1000": 40 }, + "book_depth_usd": 2500.0 + }, + "slippage_ok_100": true }, { - "pair": "SAMPLE Kalshi SENATE-WW-REP <-> SAMPLE PredictIt 'WW Senate race' (Republican)", - "best_direction": {"net_yield_c": -2.0, "net_yield_pct": -2.2}, - "executable": false + "event": "SAMPLE State WW Senate winner? - Republican", + "match_score": 0.71, + "kalshi": { + "ticker": "SAMPLE-SENATEWW-26-R", + "yes_bid_c": 80, + "yes_ask_c": 81 + }, + "predictit": { + "market": "SAMPLE WW Senate race", + "contract": "Republican", + "yes_ask_c": 83, + "no_ask_c": 10, + "position_limit_usd": 850 + }, + "best_direction": { + "direction": "YES_KALSHI__NO_PREDICTIT", + "kalshi_price_c": 81, + "predictit_price_c": 10, + "stake_c": 91, + "raw_spread_c": 9, + "net_if_yes_wins_c": -2.0, + "net_if_no_wins_c": 4.3, + "net_yield_c": -2.0, + "net_yield_pct": -2.2, + "fees": { + "kalshi_taker_c": 2, + "predictit_profit_fee_c": 8.1, + "predictit_withdrawal_drag_c": 4.05 + } + }, + "days_to_settlement": 400, + "roc_annualized_pct": -2.0, + "executable": false, + "liquidity": { + "vwap_buy_c": { "100": 81, "500": 82, "1000": 82 }, + "book_depth_usd": 2500.0 + }, + "slippage_ok_100": false }, { - "pair": "SAMPLE record with missing fields (client must tolerate this)" + "event": "SAMPLE record with missing fields (client must tolerate this)" } - ] + ], + "stats": { + "pairs_evaluated": 5, + "executable": 2, + "kalshi_markets_scanned": 5, + "predictit_contracts_scanned": 5, + "duration_ms": 0 + }, + "fetched_at": "2026-01-01T00:00:00.000Z" } From bb3097986ad323435e297454ba3806c54ecfdffb Mon Sep 17 00:00:00 2001 From: mastertyrone Date: Wed, 30 Sep 2026 10:12:07 -0600 Subject: [PATCH 12/12] Read event with pair fallback in arb feed example The feed labels rows with "event"; "pair" is the older name. The example strategy now keys and logs on event, falls back to pair, and skips rows that have neither, so no blank or "None" labels. The offline sample fixture now uses the live row shape (event, kalshi, predictit, best_direction, stats, fetched_at) with fictional values and is still marked as sample data. Adds tests for event-first, pair fallback, rows with neither skipped, and the fixture shape. --- tests/test_example_arbfeed.py | 43 +++++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/tests/test_example_arbfeed.py b/tests/test_example_arbfeed.py index e7621038..6548ca8f 100644 --- a/tests/test_example_arbfeed.py +++ b/tests/test_example_arbfeed.py @@ -100,6 +100,15 @@ def test_scan_all_limit_q(self): self.assertEqual(len(response["opportunities"]), 1) self.session.get.assert_not_called() + def test_sample_event_shape(self): + rows = self.scanner.scan(mode="all")["opportunities"] + self.assertTrue(all(row["event"].startswith("SAMPLE") for row in rows)) + self.assertEqual(rows[0]["kalshi"]["ticker"], "SAMPLE-SENATEXX-26-D") + strategy = ArbSignalStrategy( + context={"arb_signals": rows, "arb_signals_updated": time.time()} + ) + self.assertEqual(len(strategy.active_signals()), 2) + def test_scan_validation(self): with self.assertRaises(ValueError): self.scanner.scan(limit=26) @@ -361,3 +370,37 @@ def test_process_market_book(self): self.strategy.process_market_book(self.market, self.market_book) self.strategy.process_market_book(self.market, self.market_book) self.assertEqual(self.market.context["arb_signals_seen"], {"A <-> B"}) + + def test_event_first(self): + signal = { + "event": "E", + "pair": "P", + "best_direction": {"net_yield_c": 2.5}, + "executable": True, + } + self._signals([signal]) + self.assertEqual(self.strategy.active_signals(), [signal]) + self.strategy.process_market_book(self.market, self.market_book) + self.assertEqual(self.market.context["arb_signals_seen"], {"E"}) + + def test_pair_fallback(self): + signal = { + "pair": "P", + "best_direction": {"net_yield_c": 2.5}, + "executable": True, + } + self._signals([signal]) + self.assertEqual(self.strategy.active_signals(), [signal]) + self.strategy.process_market_book(self.market, self.market_book) + self.assertEqual(self.market.context["arb_signals_seen"], {"P"}) + + def test_missing_event_and_pair_skipped(self): + self._signals( + [ + {"best_direction": {"net_yield_c": 2.5}, "executable": True}, + {"event": "", "pair": None, "executable": True}, + ] + ) + self.assertEqual(self.strategy.active_signals(), []) + self.strategy.process_market_book(self.market, self.market_book) + self.assertEqual(self.market.context["arb_signals_seen"], set())