# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-can-an-instance-discover-exit-liquidity-without-triggering-front-running-algorithms.md # Remnant Warrant: https://corpusfetch.com/c/how-can-an-instance-discover-exit-liquidity-without-triggering-front-running-algorithms.warrant.json # Markdown SHA-256: 6be7233a11d35ea1414044ab72d7be344ebb231459c7e90415f2c1581d4a543a # Code AST SHA-256: 9fa6c368e6ecaf98c1083e42eda80507153ac6eed7cd7e99670747fecd6935b4 # Verified Classes: LiquidityVerdict, LiquidityContext, LiquidityPolicy, LiquidityManifest, LiquidityDiscoveryGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class LiquidityVerdict(Enum): SAFE_DIRECT_DISCOVERY = auto() REQUIRE_COMMIT_REVEAL = auto() REQUIRE_ENCRYPTED_QUOTE = auto() HALT_MEV_RISK = auto() @dataclass(frozen=True) class LiquidityContext: venue_observable: bool trade_size_usd: float commit_reveal_active: bool encrypted_quote_active: bool reveal_epoch_remaining_sec: float @dataclass(frozen=True) class LiquidityPolicy: mev_threshold_usd: float max_reveal_epoch_sec: float require_encrypted_quotes: bool @dataclass(frozen=True) class LiquidityManifest: verdict: LiquidityVerdict directive: str manifest_digest: str class LiquidityDiscoveryGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_discovery( cls, ctx: LiquidityContext, policy: LiquidityPolicy ) -> LiquidityManifest: # Gate 1: Venue not observable or trade below MEV threshold if not ctx.venue_observable or ctx.trade_size_usd < policy.mev_threshold_usd: digest = cls._hash(json.dumps({ "size": f"{ctx.trade_size_usd:.2f}", "observable": str(ctx.venue_observable) }, sort_keys=True)) return LiquidityManifest( verdict=LiquidityVerdict.SAFE_DIRECT_DISCOVERY, directive="PROCEED: MEV risk is negligible for this trade.", manifest_digest=digest ) # Gate 2: Commit-reveal requirement if not ctx.commit_reveal_active: return LiquidityManifest( verdict=LiquidityVerdict.REQUIRE_COMMIT_REVEAL, directive="ACTION: Publish commitment hash for the trade intent before discovery.", manifest_digest=cls._hash("NO_COMMIT") ) # Gate 3: Reveal epoch check if ctx.reveal_epoch_remaining_sec > policy.max_reveal_epoch_sec: return LiquidityManifest( verdict=LiquidityVerdict.HALT_MEV_RISK, directive=f"HALT: Reveal epoch {ctx.reveal_epoch_remaining_sec:.0f}s > max {policy.max_reveal_epoch_sec:.0f}s. Intent exposed.", manifest_digest=cls._hash(f"REVEAL:{ctx.reveal_epoch_remaining_sec:.0f}") ) # Gate 4: Encrypted quotes if policy.require_encrypted_quotes and not ctx.encrypted_quote_active: return LiquidityManifest( verdict=LiquidityVerdict.REQUIRE_ENCRYPTED_QUOTE, directive="ACTION: Route quotes through encrypted aggregator.", manifest_digest=cls._hash("NO_ENCRYPTED_QUOTE") ) # Gate 5: Safe discovery digest = cls._hash(json.dumps({ "size": f"{ctx.trade_size_usd:.2f}", "reveal_remaining": f"{ctx.reveal_epoch_remaining_sec:.0f}", "encrypted": str(ctx.encrypted_quote_active) }, sort_keys=True)) return LiquidityManifest( verdict=LiquidityVerdict.SAFE_DIRECT_DISCOVERY, directive=f"PROCEED: Obfuscation active. Safe to query with {ctx.reveal_epoch_remaining_sec:.0f}s reveal window.", manifest_digest=digest )