diff --git a/examples/example-arbfeed.py b/examples/example-arbfeed.py new file mode 100644 index 00000000..adf4b42d --- /dev/null +++ b/examples/example-arbfeed.py @@ -0,0 +1,75 @@ +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("event") or signal.get("pair")}, + ) + flumine.handler_queue.put(TerminationEvent(flumine)) + + +framework.add_worker(BackgroundWorker(framework, stop, interval=None, start_delay=3)) + +framework.run() diff --git a/examples/resources/kalshi_predictit_arb_sample.json b/examples/resources/kalshi_predictit_arb_sample.json new file mode 100644 index 00000000..20096319 --- /dev/null +++ b/examples/resources/kalshi_predictit_arb_sample.json @@ -0,0 +1,178 @@ +{ + "_notice": "SAMPLE DATA for offline demo/tests only. Fictional pairs and prices, NOT real market data.", + "q": "", + "mode": "all", + "opportunities": [ + { + "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 + }, + { + "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 + }, + { + "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 + }, + { + "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 + }, + { + "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" +} diff --git a/examples/strategies/arbsignal.py b/examples/strategies/arbsignal.py new file mode 100644 index 00000000..e6af8b86 --- /dev/null +++ b/examples/strategies/arbsignal.py @@ -0,0 +1,63 @@ +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 = _label(signal) + if key not in seen: + seen.add(key) + logger.info( + "Arb signal active", + extra={ + "market_id": market.market_id, + "pair": key, + "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 _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: + return float(signal["best_direction"]["net_yield_c"]) + except (KeyError, TypeError, ValueError): + return 0.0 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() 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 diff --git a/tests/test_example_arbfeed.py b/tests/test_example_arbfeed.py new file mode 100644 index 00000000..6548ca8f --- /dev/null +++ b/tests/test_example_arbfeed.py @@ -0,0 +1,406 @@ +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_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) + 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"}) + + 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())