5 posts tagged with "market-discovery"

This post covers Phase 1 of ADR-017 (Automated Market Discovery and Matching) - establishing the data types and persistence layer for the discovery system.

The Problem​

Manual market mapping is error-prone and doesn't scale. Polymarket and Kalshi list hundreds of markets; finding equivalent pairs requires:

1. Persistent storage - Track discovered markets across restarts
2. Status tracking - Pending → Approved/Rejected workflow
3. Audit trail - Record all approval decisions for compliance
4. Safety gates - Prevent automated trading without human review

Design Decisions​

CandidateStatus State Machine​

The core safety mechanism is a one-way state machine:

Pending ──┬──► Approved
          │
          └──► Rejected

Once a candidate is approved or rejected, the status is immutable. This prevents accidental re-processing or status manipulation:

impl CandidateStatus {
    pub fn can_transition_to(&self, new_status: CandidateStatus) -> bool {
        match (self, new_status) {
            (CandidateStatus::Pending, CandidateStatus::Approved) => true,
            (CandidateStatus::Pending, CandidateStatus::Rejected) => true,
            // Once approved or rejected, status is final
            (CandidateStatus::Approved, _) => false,
            (CandidateStatus::Rejected, _) => false,
            _ => false,
        }
    }
}

Semantic Warnings​

Markets that appear similar may have different settlement criteria. The CandidateMatch struct includes a semantic_warnings field that Phase 2's matcher will populate:

pub struct CandidateMatch {
    pub semantic_warnings: Vec<String>,  // e.g., "Settlement timing differs"
    // ...
}

Approval will require explicit acknowledgment of these warnings (FR-MD-003).

SQLite Storage​

We chose SQLite over PostgreSQL for the discovery cache because:

1. Single-tenant - Discovery runs locally per operator
2. Portable - No external dependencies for development
3. Atomic - Transactions prevent partial state

Schema design separates markets from candidates:

-- Discovered markets (one per platform/id combination)
CREATE TABLE discovered_markets (
    id TEXT PRIMARY KEY,
    platform TEXT NOT NULL,
    platform_id TEXT NOT NULL,
    title TEXT NOT NULL,
    -- ...
    UNIQUE(platform, platform_id)
);

-- Candidate matches (references two markets)
CREATE TABLE candidates (
    id TEXT PRIMARY KEY,
    polymarket_id TEXT NOT NULL,
    kalshi_id TEXT NOT NULL,
    similarity_score REAL NOT NULL,
    status TEXT NOT NULL DEFAULT 'Pending',
    -- ...
);

-- Audit log for compliance
CREATE TABLE audit_log (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    timestamp TEXT NOT NULL,
    action TEXT NOT NULL,
    candidate_id TEXT NOT NULL,
    details TEXT NOT NULL  -- Full JSON context
);

Parameterized Queries​

All SQL uses the params![] macro to prevent injection:

conn.execute(
    "UPDATE candidates SET status = ?1, updated_at = ?2 WHERE id = ?3",
    params![status_str, now, id.to_string()],
)?;

Test Coverage​

Phase 1 includes 12 tests covering:

Key safety test:

#[test]
fn test_candidate_status_transitions() {
    // Once approved, cannot transition to any other status
    assert!(!CandidateStatus::Approved.can_transition_to(CandidateStatus::Pending));
    assert!(!CandidateStatus::Approved.can_transition_to(CandidateStatus::Rejected));
}

What's Next​

Phase 2 will implement the text matching engine:

• TextNormalizer - Lowercase, remove punctuation, tokenize
• SimilarityScorer - Jaccard (0.6 weight) + Levenshtein (0.4 weight)
• Semantic warning detection for settlement differences

Council Review​

Phase 1 passed council verification with confidence 0.88. Key findings:

• ✅ Human-in-the-loop enforced via CandidateStatus state machine
• ✅ Audit logging captures all required fields
• ✅ No SQL injection (all parameterized queries)
• ✅ No unsafe code

Implementation: arbiter-engine/src/discovery/ | Issues: #41, #42 | ADR: 017

This post covers Phase 2 of ADR-017 - implementing the text similarity matching engine that powers automated market discovery between Polymarket and Kalshi.

The Problem​

Phase 1 established the data types and storage layer. Now we need to actually find matching markets across platforms. The challenge:

1. Fuzzy matching - Market titles differ in phrasing ("Will Trump win?" vs "Trump wins 2024?")
2. False positives - Similar titles may have different settlement criteria
3. Scalability - Must compare thousands of markets efficiently

Algorithm Design​

Combined Similarity Scoring​

We use a weighted combination of two complementary algorithms:

score = 0.6 × Jaccard + 0.4 × Levenshtein

Jaccard similarity (0.6 weight) measures token set overlap:

let intersection = set_a.intersection(&set_b).count();
let union = set_a.union(&set_b).count();
jaccard = intersection / union

This captures semantic similarity when words are reordered.

Levenshtein similarity (0.4 weight) measures edit distance:

let distance = levenshtein(&norm_a, &norm_b);
levenshtein_sim = 1.0 - (distance / max_length)

This catches typos and minor variations.

Text Normalization​

Before comparison, titles are normalized:

impl TextNormalizer {
    pub fn normalize(&self, text: &str) -> String {
        // 1. Lowercase
        // 2. Replace punctuation with spaces
        // 3. Collapse whitespace
    }

    pub fn tokenize(&self, text: &str) -> Vec<String> {
        // 4. Split into words
        // 5. Filter stop words (a, an, the, will, be, ...)
    }
}

Example: "Will Bitcoin reach $100k?" → ["bitcoin", "reach", "100k"]

Pre-Filtering​

Before scoring, candidates are filtered to reduce false positives:

Semantic Warning Detection (FR-MD-008)​

Even similar titles may have different settlement criteria. We detect and flag:

Conditional language mismatches:

Polymarket: "Will Fed announce rate cut?"
Kalshi:     "Will Fed cut rates?"
⚠️ Warning: Settlement trigger mismatch - one market references 'announce'

Resolution source differences:

Polymarket resolution: "Associated Press"
Kalshi resolution:     "Official FEC results"
⚠️ Warning: Resolution source differs

Expiration differences:

⚠️ Warning: Expiration differs by 3 day(s)

These warnings flow to the human reviewer (FR-MD-003) for acknowledgment before approval.

Implementation​

SimilarityScorer​

pub struct SimilarityScorer {
    jaccard_weight: f64,      // 0.6
    levenshtein_weight: f64,  // 0.4
    threshold: f64,           // 0.6
    normalizer: TextNormalizer,
    pre_filter: PreFilterConfig,
}

impl SimilarityScorer {
    pub fn find_matches(
        &self,
        market: &DiscoveredMarket,
        candidates: &[DiscoveredMarket],
    ) -> Vec<CandidateMatch> {
        candidates.iter()
            .filter(|c| c.platform != market.platform)  // Cross-platform only
            .filter(|c| self.passes_pre_filter(market, c))
            .filter_map(|c| {
                let score = self.score(&market.title, &c.title);
                if score >= self.threshold {
                    let warnings = self.detect_warnings(market, c);
                    Some(CandidateMatch::new(/*...*/).with_warnings(warnings))
                } else {
                    None
                }
            })
            .collect()
    }
}

Match Reason Classification​

let match_reason = if score >= 0.95 {
    MatchReason::ExactTitle
} else {
    MatchReason::HighTextSimilarity { score: (score * 100.0) as u32 }
};

Test Coverage​

Phase 2 adds 10 tests (22 total for discovery module):

Key test:

#[test]
fn test_semantic_warning_announcement() {
    let scorer = SimilarityScorer::default();

    let poly = create_market(Platform::Polymarket, "Will Fed announce rate cut?");
    let kalshi = create_market(Platform::Kalshi, "Will Fed cut rates?");

    let warnings = scorer.detect_warnings(&poly, &kalshi);
    assert!(warnings.iter().any(|w| w.contains("announce")));
}

What's Next​

Phase 3 will implement the API clients:

• Polymarket Gamma API client (FR-MD-006)
• Kalshi /v2/markets API client (FR-MD-007)
• Rate limiting and pagination

Council Review​

Phase 2 passed council verification with confidence 0.85. Key findings:

• No unsafe code
• Human-in-the-loop preserved (find_matches returns candidates, not verified mappings)
• Semantic warnings properly flag settlement differences
• All tests passing (22 total)

Implementation: arbiter-engine/src/discovery/{normalizer,matcher}.rs | Issue: #43 | ADR: 017

This post covers Phase 3 of ADR-017 - implementing the API clients that fetch market listings from Polymarket and Kalshi for automated discovery.

The Problem​

Phase 1 and 2 established storage and matching. Now we need data sources:

1. Polymarket - Gamma API at gamma-api.polymarket.com
2. Kalshi - Trade API at api.elections.kalshi.com/trade-api/v2

Both APIs have:

• Pagination (different styles)
• Rate limits (different thresholds)
• Different response schemas

Design: DiscoveryClient Trait​

We define a common trait for both platforms:

#[async_trait]
pub trait DiscoveryClient: Send + Sync {
    async fn list_markets(
        &self,
        limit: Option<u32>,
        cursor: Option<&str>,
    ) -> Result<DiscoveryPage, DiscoveryError>;

    fn platform_name(&self) -> &'static str;
}

This allows the scanner (Phase 4) to enumerate markets from either platform interchangeably.

Rate Limiting​

Both APIs have rate limits we must respect:

We implement a token bucket rate limiter:

struct RateLimiter {
    tokens: AtomicU64,
    last_refill: Mutex<Instant>,
    max_tokens: u32,
}

impl RateLimiter {
    async fn acquire(&self) -> Option<Duration> {
        // Refill tokens based on elapsed time
        let elapsed = last.elapsed();
        let refill = (elapsed.as_secs_f64() / 60.0 * max_tokens) as u64;

        // Try to consume a token
        if tokens > 0 {
            tokens -= 1;
            return None; // Success
        }

        // Return wait time
        Some(Duration::from_secs_f64(60.0 / max_tokens))
    }
}

If rate limited, we return DiscoveryError::RateLimited with the retry time.

Pagination Strategies​

Polymarket: Offset-based​

GET /markets?limit=100&offset=0
GET /markets?limit=100&offset=100
...

We use the offset as the cursor, incrementing by page size.

Kalshi: Cursor-based​

GET /markets?limit=100&status=open
→ { markets: [...], cursor: "abc123" }

GET /markets?limit=100&cursor=abc123
→ { markets: [...], cursor: null }

We pass through the cursor directly.

Response Mapping​

Each API returns different schemas that we map to DiscoveredMarket:

Polymarket Gamma API​

struct GammaMarket {
    condition_id: String,    // → platform_id
    question: String,        // → title
    outcomes: String,        // JSON array → outcomes
    end_date: String,        // → expiration
    volume_24hr: f64,        // → volume_24h
    active: bool,            // Filter: skip if false
    closed: bool,            // Filter: skip if true
}

Kalshi Markets API​

struct KalshiMarket {
    ticker: String,          // → platform_id
    title: String,           // → title
    expiration_time: String, // → expiration
    volume_24h: i64,         // Cents → dollars
    status: String,          // Filter: only "open"/"active"
}

Key transformations:

• Kalshi volume is in cents, converted to dollars (/ 100.0)
• Inactive/closed markets are filtered out before returning
• Missing fields use sensible defaults

Error Handling​

pub enum DiscoveryError {
    Http(reqwest::Error),           // Network failures
    Parse(String),                  // JSON parsing
    RateLimited { retry_after_secs: u64 },  // 429 responses
    ApiError { status: u16, message: String },  // Other HTTP errors
}

The scanner (Phase 4) can handle these appropriately - retrying on rate limits, logging API errors.

Test Strategy​

We use wiremock for HTTP mocking:

#[tokio::test]
async fn test_list_markets_success() {
    let mock_server = MockServer::start().await;

    Mock::given(method("GET"))
        .and(path("/markets"))
        .respond_with(ResponseTemplate::new(200)
            .set_body_json(mock_response()))
        .mount(&mock_server)
        .await;

    let client = GammaApiClient::with_base_url(&mock_server.uri());
    let page = client.list_markets(Some(10), None).await.unwrap();

    assert_eq!(page.markets.len(), 2);
}

Test Coverage​

Phase 3 adds 8 tests (30 total for discovery):

What's Next​

Phase 4 will implement the scanner and approval workflow:

• DiscoveryScannerActor for periodic discovery runs
• ApprovalWorkflow for human review (FR-MD-003)
• Integration with MappingManager.verify_mapping()

Council Review​

Phase 3 passed council verification with confidence 0.87. Key findings:

• No unsafe code
• Proper rate limiting prevents API abuse
• 30-second timeout prevents hanging
• No credentials hardcoded
• Closed/inactive markets filtered out

Implementation: arbiter-engine/src/market/discovery_client/ | Issues: #44, #45 | ADR: 017

This post covers Phase 4 of ADR-017 - the scanner actor for periodic discovery and the safety-critical human approval workflow.

The Problem​

Phase 1-3 established storage, matching, and API clients. Now we need:

1. Automated Discovery - Periodic scanning of both platforms
2. Human Approval - Safety gate preventing automated mappings from entering trading

This phase implements FR-MD-003 (human confirmation required) and FR-MD-004 (auto-discover markets).

Safety-First Design​

FR-MD-003 is SAFETY CRITICAL. The approval workflow enforces:

1. Warning Acknowledgment - Cannot approve candidates with semantic warnings without explicit acknowledgment
2. Audit Logging - All decisions logged with full context
3. MappingManager Integration - Approved candidates go through existing safety gate

pub fn approve(&self, id: Uuid, acknowledge_warnings: bool) -> Result<Uuid, ApprovalError> {
    let candidate = self.get_candidate(id)?;

    // SAFETY CHECK: Require warning acknowledgment if warnings exist
    if !candidate.semantic_warnings.is_empty() && !acknowledge_warnings {
        return Err(ApprovalError::WarningsNotAcknowledged);
    }

    // Create verified mapping through the existing safety gate
    let mapping_id = {
        let mut manager = self.mapping_manager.lock().unwrap();
        let id = manager.propose_mapping(/*...*/);
        manager.verify_mapping(id);  // MappingManager safety gate
        id
    };

    // Update status and log decision
    // ...
}

Scanner Actor​

The DiscoveryScannerActor implements the Actor trait for periodic discovery:

pub enum ScannerMsg {
    Scan,           // Trigger a scan
    ForceRefresh,   // Ignore cache
    Stop,           // Graceful shutdown
    GetStatus(tx),  // Query status
}

Deduplication​

The scanner prevents duplicate candidates:

async fn is_duplicate_candidate(&self, candidate: &CandidateMatch) -> Result<bool, ScanError> {
    let storage = self.storage.lock().await;

    // Check pending candidates
    let pending = storage.query_candidates_by_status(CandidateStatus::Pending)?;
    for existing in pending {
        if existing.polymarket.platform_id == candidate.polymarket.platform_id
            && existing.kalshi.platform_id == candidate.kalshi.platform_id
        {
            return Ok(true);
        }
    }

    // Also check approved candidates
    let approved = storage.query_candidates_by_status(CandidateStatus::Approved)?;
    // ...
}

Scan Flow​

1. Fetch markets from Polymarket (with pagination)
2. Fetch markets from Kalshi (with cursor pagination)
3. Store all markets in SQLite
4. Run similarity matching
5. Deduplicate against existing candidates
6. Store new candidates with Pending status

Approval Workflow​

The ApprovalWorkflow provides the human interface:

// List candidates awaiting review
let pending = workflow.list_pending()?;

// Approve (must acknowledge warnings if present)
let mapping_id = workflow.approve(candidate_id, true)?;

// Reject with reason (required)
workflow.reject(candidate_id, "Different settlement criteria")?;

Rejection Requires Reason​

To maintain audit trail quality, rejections require a non-empty reason:

pub fn reject(&self, id: Uuid, reason: &str) -> Result<(), ApprovalError> {
    if reason.trim().is_empty() {
        return Err(ApprovalError::ReasonRequired);
    }
    // ...
}

Audit Trail​

Every decision is logged with full context:

let entry = AuditLogEntry {
    timestamp: Utc::now(),
    action: AuditAction::Approve,  // or Reject
    candidate_id: id,
    polymarket_id: candidate.polymarket.platform_id.clone(),
    kalshi_id: candidate.kalshi.platform_id.clone(),
    similarity_score: candidate.similarity_score,
    semantic_warnings: candidate.semantic_warnings.clone(),
    acknowledged_warnings: acknowledge_warnings,
    reason: None,  // or Some("...") for rejections
    session_id: self.session_id.clone(),
};
storage.append_audit_log(&entry)?;

Test Coverage​

Phase 4 adds 10 tests (40 total for discovery):

Critical Safety Test​

#[test]
fn test_approve_requires_warning_acknowledgment() {
    // Add candidate WITH warnings
    let candidate = setup_candidate(&storage, true);
    let workflow = ApprovalWorkflow::new(storage, mapping_manager);

    // Try to approve WITHOUT acknowledging warnings - MUST FAIL
    let result = workflow.approve(candidate.id, false);
    assert!(result.is_err(), "SAFETY VIOLATION: Should require warning acknowledgment");

    match result {
        Err(ApprovalError::WarningsNotAcknowledged) => {
            // Correct error type
        }
        Ok(_) => panic!("SAFETY VIOLATION: Approved without acknowledging warnings!"),
        // ...
    }
}

What's Next​

Phase 5 will implement CLI integration:

• --discover-markets - Trigger discovery scan
• --list-candidates - List pending/approved/rejected
• --approve-candidates - Approve by ID
• --reject-candidates - Reject with reason

Council Review​

Phase 4 passed council verification with confidence 0.91 (Safety focus). Key findings:

• FR-MD-003 enforcement verified
• Warning acknowledgment required
• Audit logging on all decisions
• Integration with MappingManager.verify_mapping() confirmed
• Deduplication prevents duplicate reviews

Implementation: arbiter-engine/src/discovery/{scanner,approval}.rs | Issues: #46, #47 | ADR: 017

This post covers Phase 5, the final phase of ADR-017 - CLI command integration for the discovery workflow.

The Problem​

Phases 1-4 built the complete discovery infrastructure:

• Storage and data types
• Text similarity matching
• API clients for both platforms
• Scanner actor and approval workflow

But operators had no way to interact with this system. Phase 5 bridges that gap.

CLI Commands​

Four commands enable the human-in-the-loop workflow:

# Trigger discovery scan
cargo run --features discovery -- --discover-markets

# List candidates by status
cargo run --features discovery -- --list-candidates --status pending
cargo run --features discovery -- --list-candidates --status approved
cargo run --features discovery -- --list-candidates --status rejected
cargo run --features discovery -- --list-candidates --status all

# Approve a candidate (with optional warning acknowledgment)
cargo run --features discovery -- --approve-candidate <uuid>
cargo run --features discovery -- --approve-candidate <uuid> --acknowledge-warnings

# Reject with required reason
cargo run --features discovery -- --reject-candidate <uuid> --reason "Different settlement criteria"

Testable Command Handlers​

The CLI handlers are separated from main.rs into src/discovery/cli.rs for testability:

pub struct DiscoveryCli {
    storage: Arc<Mutex<CandidateStorage>>,
    mapping_manager: Arc<Mutex<MappingManager>>,
    config: DiscoveryCliConfig,
}

impl DiscoveryCli {
    pub fn list_candidates(&self, status: Option<CandidateStatus>) -> CliResult { ... }
    pub fn approve_candidate(&self, id: Uuid, acknowledge_warnings: bool) -> CliResult { ... }
    pub fn reject_candidate(&self, id: Uuid, reason: &str) -> CliResult { ... }
}

This separation allows comprehensive unit testing without spawning the full async runtime.

Safety Enforcement at CLI Layer​

The CLI layer preserves FR-MD-003 safety guarantees:

pub fn approve_candidate(&self, id: Uuid, acknowledge_warnings: bool) -> CliResult {
    let workflow = ApprovalWorkflow::new(...);

    match workflow.approve(id, acknowledge_warnings) {
        Ok(mapping_id) => CliResult::Success(format!(
            "Candidate {} approved. Verified mapping ID: {}", id, mapping_id
        )),
        Err(ApprovalError::WarningsNotAcknowledged) => CliResult::Error(
            "Cannot approve: candidate has semantic warnings. \
             Use --acknowledge-warnings to proceed.".to_string()
        ),
        // ... other error handling
    }
}

Error messages guide operators to the correct action.

Feature Gate Error Handling​

When the discovery feature is not enabled, helpful error messages are shown:

#[cfg(not(feature = "discovery"))]
{
    if is_discovery_command {
        eprintln!("Discovery commands require the 'discovery' feature.");
        eprintln!("   Build with: cargo build --features discovery");
        eprintln!("   Run with:   cargo run --features discovery -- --discover-markets");
        return Ok(());
    }
}

Test Coverage​

Phase 5 adds 8 tests (48 total for discovery, 377 overall):

ADR-017 Complete​

With Phase 5, ADR-017 is fully implemented:

Council Review​

Phase 5 passed council verification with confidence 0.95 (Safety focus). Key findings:

• FR-MD-003 enforcement verified at CLI layer
• Warning acknowledgment required for candidates with semantic warnings
• Rejection requires non-empty reason for audit trail
• Clear error messages guide operators
• Feature gate prevents confusion when feature disabled
• No code path bypasses human review

Implementation: arbiter-engine/src/discovery/cli.rs | Issue: #48 | ADR: 017