19 posts tagged with "arbiter-bot"

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

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

    // Use existing safety gate (FR-MD-003)
    let mut manager = self.mapping_manager.lock().unwrap();
    let mapping_id = manager.propose_mapping(/*...*/);
    manager.verify_mapping(mapping_id);

    // Audit log for compliance
    self.storage.log_decision(/*...*/)?;
    Ok(())
}

pub struct SimilarityScorer {
    jaccard_weight: f64,      // 0.6
    levenshtein_weight: f64,  // 0.4
    threshold: f64,           // 0.6
}

impl SimilarityScorer {
    pub fn find_matches(&self, market: &DiscoveredMarket, candidates: &[DiscoveredMarket])
        -> Vec<CandidateMatch>
    {
        candidates.iter()
            .filter(|c| self.pre_filter(market, c))
            .filter_map(|c| {
                let score = self.combined_score(&market.title, &c.title);
                if score >= self.threshold {
                    Some(CandidateMatch::new(market.clone(), c.clone(), score))
                } else {
                    None
                }
            })
            .collect()
    }
}

impl Actor for DiscoveryScannerActor {
    type Message = ScannerMsg;

    async fn handle(&mut self, message: Self::Message) -> Result<(), ActorError> {
        match message {
            ScannerMsg::Scan => {
                let poly_markets = self.fetch_all_markets(&*self.polymarket_client).await?;
                let kalshi_markets = self.fetch_all_markets(&*self.kalshi_client).await?;

                // Store markets
                for market in &poly_markets {
                    self.storage.lock().await.upsert_market(market)?;
                }

                // Find candidates
                for poly_market in &poly_markets {
                    let matches = self.scorer.find_matches(poly_market, &kalshi_markets);
                    for candidate in matches {
                        if !self.is_duplicate_candidate(&candidate).await? {
                            self.storage.lock().await.insert_candidate(&candidate)?;
                        }
                    }
                }
            }
            // ...
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum KalshiEnvironment {
    #[default]
    Production,
    Demo,
}

impl KalshiEnvironment {
    pub fn api_base_url(&self) -> &'static str {
        match self {
            KalshiEnvironment::Production => "https://trading-api.kalshi.com",
            KalshiEnvironment::Demo => "https://demo-api.kalshi.co",
        }
    }

    pub fn websocket_url(&self) -> &'static str {
        match self {
            KalshiEnvironment::Production => "wss://trading-api.kalshi.com/trade-api/v2/ws",
            KalshiEnvironment::Demo => "wss://demo-api.kalshi.co/trade-api/v2/ws",
        }
    }
}

let (kalshi_key_var, kalshi_priv_var) = if args.kalshi_demo {
    ("KALSHI_DEMO_KEY_ID", "KALSHI_DEMO_PRIVATE_KEY")
} else {
    ("KALSHI_KEY_ID", "KALSHI_PRIVATE_KEY")
};

impl KalshiClient {
    pub fn new(key_id: String, private_key_pem: &str, dry_run: bool) -> Result<Self, String> {
        // Default to production
        Self::with_environment(key_id, private_key_pem, dry_run, KalshiEnvironment::Production)
    }

    pub fn with_environment(
        key_id: String,
        private_key_pem: &str,
        dry_run: bool,
        environment: KalshiEnvironment,
    ) -> Result<Self, String> {
        // ... initialization with environment-specific URLs
    }
}

#[derive(Parser, Debug)]
struct Args {
    /// Use Kalshi demo environment instead of production
    #[arg(long, default_value_t = false)]
    kalshi_demo: bool,
    // ...
}

#[test]
fn test_production_urls() {
    let env = KalshiEnvironment::Production;
    assert_eq!(env.api_base_url(), "https://trading-api.kalshi.com");
    assert_eq!(env.websocket_url(), "wss://trading-api.kalshi.com/trade-api/v2/ws");
}

#[test]
fn test_demo_urls() {
    let env = KalshiEnvironment::Demo;
    assert_eq!(env.api_base_url(), "https://demo-api.kalshi.co");
    assert_eq!(env.websocket_url(), "wss://demo-api.kalshi.co/trade-api/v2/ws");
}

#[test]
fn test_default_is_production() {
    assert_eq!(KalshiEnvironment::default(), KalshiEnvironment::Production);
}

// Future: Polymarket demo (if they add one)
pub enum PolymarketEnvironment {
    #[default]
    Production,
    Demo,  // Hypothetical
}

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,
        }
    }
}

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

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

#[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));
}

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

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

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, ...)
    }
}

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()
    }
}

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

#[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")));
}

#[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;
}

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))
    }
}

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
}

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

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
}

#[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);
}

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
    // ...
}

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

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)?;
    // ...
}

// 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")?;

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

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]
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!"),
        // ...
    }
}

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 { ... }
}

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
    }
}

#[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(());
    }
}

// WRONG: Race condition
let nonce = self.nonce + 1;
self.nonce = nonce; // Another thread could read same value

pub struct NonceManager {
    nonces: RwLock<HashMap<String, Arc<AtomicU64>>>,
}

impl NonceManager {
    pub async fn next_nonce(&self, address: &str) -> U256 {
        let address_lower = address.to_lowercase();

        // Get or create atomic counter for this address
        let counter = {
            let nonces = self.nonces.read().await;
            if let Some(counter) = nonces.get(&address_lower) {
                counter.clone()
            } else {
                drop(nonces);
                let mut nonces = self.nonces.write().await;
                let counter = Arc::new(AtomicU64::new(
                    Utc::now().timestamp_millis() as u64
                ));
                nonces.insert(address_lower.clone(), counter.clone());
                counter
            }
        };

        // Atomic increment - guaranteed unique
        U256::from(counter.fetch_add(1, Ordering::SeqCst))
    }
}

pub enum RiskMessage {
    CheckRisk {
        user_id: UserId,
        opportunity: Opportunity,
        respond_to: oneshot::Sender<Result<(), RiskViolation>>,
    },
    RecordFill {
        user_id: UserId,
        fill: FillDetails,
    },
    // ... other messages
}

impl RiskManagerActor {
    pub async fn run(mut self) {
        while let Some(msg) = self.receiver.recv().await {
            match msg {
                RiskMessage::CheckRisk { user_id, opportunity, respond_to } => {
                    let result = self.check_risk(&user_id, &opportunity);
                    let _ = respond_to.send(result);
                }
                RiskMessage::RecordFill { user_id, fill } => {
                    self.record_fill(&user_id, &fill);
                }
            }
        }
    }
}

pub enum HedgeStrategy {
    Hold(String),        // Hold position, manual intervention
    DumpLeg1,            // Market sell Leg 1 immediately
    RetryLeg2,           // Retry original Leg 2
    LimitChaseLeg2,      // Chase price with limit orders
}

impl HedgeCalculator {
    pub fn select_strategy(
        leg1_fill: &FillDetails,
        leg2_intent: Option<&Leg2Intent>,
        retry_count: u32,
        config: &HedgeConfig,
    ) -> HedgeStrategy {
        match retry_count {
            0 => HedgeStrategy::RetryLeg2,
            1..=2 => HedgeStrategy::LimitChaseLeg2,
            _ if config.allow_market_fallback => HedgeStrategy::DumpLeg1,
            _ => HedgeStrategy::Hold("Max retries exceeded".into()),
        }
    }
}

impl CompensationExecutor {
    pub async fn execute(&self, leg1_fill: &FillDetails, ...) -> CompensationResult {
        let mut retry_count = 0;

        loop {
            let strategy = HedgeCalculator::select_strategy(..., retry_count, ...);
            let hedge_order = HedgeCalculator::calculate(&strategy, leg1_fill);

            match self.execute_hedge_order(&hedge_order).await {
                Ok(fill) => return CompensationResult::Success(fill),
                Err(_) if retry_count < self.config.max_retries => {
                    retry_count += 1;
                    continue;
                }
                Err(e) => return CompensationResult::Failed { reason: e, ... },
            }
        }
    }
}

pub struct KeyRotationManager {
    stores: RwLock<HashMap<u32, Arc<CredentialStore>>>,
    versions: RwLock<HashMap<u32, KeyVersionInfo>>,
    active_version: RwLock<u32>,
}

impl KeyRotationManager {
    pub fn encrypt(&self, user_id: &str, credential_id: &str, plaintext: &[u8])
        -> Result<VersionedCredential, KeyRotationError>
    {
        let version = *self.active_version.read().unwrap();
        let store = self.stores.read().unwrap()
            .get(&version).cloned()
            .ok_or(KeyRotationError::NoKeysAvailable)?;

        let encrypted = store.encrypt(user_id, plaintext)?;

        Ok(VersionedCredential {
            key_version: version,
            encrypted,
            user_id: user_id.to_string(),
        })
    }

    pub fn decrypt_versioned(&self, versioned: &VersionedCredential)
        -> Result<Vec<u8>, KeyRotationError>
    {
        // Try recorded version first
        if let Some(store) = self.stores.read().unwrap().get(&versioned.key_version) {
            if let Ok(plaintext) = store.decrypt(&versioned.user_id, &versioned.encrypted) {
                return Ok(plaintext);
            }
        }

        // Try other active versions (migration fallback)
        for (&version, info) in self.versions.read().unwrap().iter() {
            if version == versioned.key_version || !info.active_for_decrypt {
                continue;
            }
            // ... try decrypt with other versions
        }

        Err(KeyRotationError::NoKeysAvailable)
    }
}