Spaces:
Paused
Paused
| #!/usr/bin/env python3 | |
| """ | |
| AirMicroDrip Audit Integration | |
| Connects the audit framework to the AirMicroDrip perpetual futures system. | |
| Provides continuous monitoring, health checks, and compliance verification. | |
| """ | |
| import os | |
| import sys | |
| from datetime import datetime | |
| from typing import Dict, Any, Optional | |
| # Add parent directory to path for audit_framework import | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| from audit_framework import ( | |
| AuditFramework, | |
| AuditCategory, | |
| AuditSeverity, | |
| AuditStatus, | |
| ) | |
| class AirMicroDripAuditor: | |
| """Audit wrapper for AirMicroDrip systems.""" | |
| def __init__(self, db_path: str = "airmicrodrip_audit.db"): | |
| self.audit = AuditFramework() | |
| self.db_path = db_path | |
| self._register_airmicrodrip_checks() | |
| def _register_airmicrodrip_checks(self): | |
| """Register AirMicroDrip-specific audit checks.""" | |
| from audit_framework import AuditCheck | |
| extra_checks = [ | |
| AuditCheck( | |
| check_id="amd_001", | |
| name="Liquidity Provider Health", | |
| description="Verify at least one active LLM inference provider", | |
| category=AuditCategory.AVAILABILITY, | |
| severity=AuditSeverity.HIGH, | |
| ), | |
| AuditCheck( | |
| check_id="amd_002", | |
| name="Synthetic Liquidity Depth", | |
| description="Verify total synthetic liquidity exceeds minimum threshold", | |
| category=AuditCategory.ACCURACY, | |
| severity=AuditSeverity.HIGH, | |
| ), | |
| AuditCheck( | |
| check_id="amd_003", | |
| name="Perpetual Engine Consistency", | |
| description="Verify mark prices and index prices are within tolerance", | |
| category=AuditCategory.ACCURACY, | |
| severity=AuditSeverity.CRITICAL, | |
| ), | |
| AuditCheck( | |
| check_id="amd_004", | |
| name="Funding Rate Bounds", | |
| description="Verify funding rates are within configured min/max", | |
| category=AuditCategory.ACCURACY, | |
| severity=AuditSeverity.MEDIUM, | |
| ), | |
| AuditCheck( | |
| check_id="amd_005", | |
| name="Liquidation Backlog", | |
| description="Verify no positions are stuck in liquidation queue", | |
| category=AuditCategory.INTEGRITY, | |
| severity=AuditSeverity.CRITICAL, | |
| ), | |
| AuditCheck( | |
| check_id="amd_006", | |
| name="Order Book Spread", | |
| description="Verify bid-ask spread is within acceptable range", | |
| category=AuditCategory.PERFORMANCE, | |
| severity=AuditSeverity.MEDIUM, | |
| ), | |
| ] | |
| for check in extra_checks: | |
| self.audit.checks[check.check_id] = check | |
| def check_liquidity_providers(self, registry) -> Dict[str, Any]: | |
| """Run liquidity provider health check.""" | |
| providers = registry.get_all_providers(status="active") | |
| if not providers: | |
| return { | |
| "status": AuditStatus.FAILED, | |
| "message": "No active liquidity providers", | |
| "details": {"active_count": 0}, | |
| } | |
| return { | |
| "status": AuditStatus.PASSED, | |
| "message": f"{len(providers)} active liquidity providers", | |
| "details": {"active_count": len(providers)}, | |
| } | |
| def check_liquidity_depth(self, converter) -> Dict[str, Any]: | |
| """Run synthetic liquidity depth check.""" | |
| total = converter.get_total_liquidity() | |
| total_usd = total.get("total_usd", 0.0) | |
| min_liquidity = float(os.environ.get("MIN_LIQUIDITY_USD", 10000.0)) | |
| if total_usd < min_liquidity: | |
| return { | |
| "status": AuditStatus.FAILED, | |
| "message": f"Total liquidity ${total_usd:.2f} below minimum ${min_liquidity:.2f}", | |
| "details": {"total_usd": total_usd, "minimum": min_liquidity}, | |
| } | |
| return { | |
| "status": AuditStatus.PASSED, | |
| "message": f"Total liquidity ${total_usd:.2f} above minimum", | |
| "details": {"total_usd": total_usd, "by_market": total.get("by_market", {})}, | |
| } | |
| def check_mark_price_consistency(self, trading_engine, tolerance: float = 0.02) -> Dict[str, Any]: | |
| """Verify mark prices are close to index prices.""" | |
| inconsistent = [] | |
| for market, state in trading_engine.market_states.items(): | |
| if state.index_price == 0: | |
| continue | |
| deviation = abs(state.mark_price - state.index_price) / state.index_price | |
| if deviation > tolerance: | |
| inconsistent.append({ | |
| "market": market, | |
| "mark": state.mark_price, | |
| "index": state.index_price, | |
| "deviation": deviation, | |
| }) | |
| if inconsistent: | |
| return { | |
| "status": AuditStatus.FAILED, | |
| "message": f"{len(inconsistent)} market(s) with price deviation > {tolerance:.1%}", | |
| "details": {"inconsistent": inconsistent}, | |
| } | |
| return { | |
| "status": AuditStatus.PASSED, | |
| "message": "Mark prices consistent with index prices", | |
| "details": {"markets_checked": len(trading_engine.market_states)}, | |
| } | |
| def check_funding_rate_bounds(self, funding_engine) -> Dict[str, Any]: | |
| """Verify funding rates within bounds.""" | |
| from funding_rate_engine import FUNDING_CONFIG | |
| out_of_bounds = [] | |
| for market in funding_engine.trading_engine.market_states: | |
| rate = funding_engine.calculate_funding_rate(market) | |
| if rate < FUNDING_CONFIG["min_funding_rate"] or rate > FUNDING_CONFIG["max_funding_rate"]: | |
| out_of_bounds.append({"market": market, "rate": rate}) | |
| if out_of_bounds: | |
| return { | |
| "status": AuditStatus.FAILED, | |
| "message": f"{len(out_of_bounds)} funding rate(s) out of bounds", | |
| "details": {"out_of_bounds": out_of_bounds}, | |
| } | |
| return { | |
| "status": AuditStatus.PASSED, | |
| "message": "All funding rates within bounds", | |
| "details": {"markets_checked": len(funding_engine.trading_engine.market_states)}, | |
| } | |
| def check_liquidation_backlog(self, liq_system) -> Dict[str, Any]: | |
| """Check for stuck liquidations.""" | |
| at_risk = liq_system.get_at_risk_positions() | |
| if len(at_risk) > 10: | |
| return { | |
| "status": AuditStatus.WARNING, | |
| "message": f"{len(at_risk)} positions at risk — possible backlog", | |
| "details": {"at_risk_count": len(at_risk)}, | |
| } | |
| return { | |
| "status": AuditStatus.PASSED, | |
| "message": f"Liquidation queue healthy ({len(at_risk)} at risk)", | |
| "details": {"at_risk_count": len(at_risk)}, | |
| } | |
| def check_orderbook_spread(self, trading_engine, max_spread_bps: float = 50.0) -> Dict[str, Any]: | |
| """Verify bid-ask spreads are within tolerance.""" | |
| wide_spreads = [] | |
| for market, ob in trading_engine.order_books.items(): | |
| best_bid = ob.get_best_bid() | |
| best_ask = ob.get_best_ask() | |
| if best_bid and best_ask and best_bid > 0: | |
| spread_bps = ((best_ask - best_bid) / best_bid) * 10000 | |
| if spread_bps > max_spread_bps: | |
| wide_spreads.append({"market": market, "spread_bps": spread_bps}) | |
| if wide_spreads: | |
| return { | |
| "status": AuditStatus.WARNING, | |
| "message": f"{len(wide_spreads)} market(s) with wide spread", | |
| "details": {"wide_spreads": wide_spreads}, | |
| } | |
| return { | |
| "status": AuditStatus.PASSED, | |
| "message": "Order book spreads within tolerance", | |
| "details": {"markets_checked": len(trading_engine.order_books)}, | |
| } | |
| def run_airmicrodrip_audit( | |
| self, | |
| registry=None, | |
| converter=None, | |
| trading_engine=None, | |
| funding_engine=None, | |
| liq_system=None, | |
| ) -> Dict[str, Any]: | |
| """Run the full AirMicroDrip audit suite.""" | |
| ctx: Dict[str, Any] = {} | |
| if registry: | |
| ctx["liquidity_providers"] = self.check_liquidity_providers(registry) | |
| if converter: | |
| ctx["liquidity_depth"] = self.check_liquidity_depth(converter) | |
| if trading_engine: | |
| ctx["price_consistency"] = self.check_mark_price_consistency(trading_engine) | |
| if funding_engine: | |
| ctx["funding_bounds"] = self.check_funding_rate_bounds(funding_engine) | |
| if liq_system: | |
| ctx["liquidation_backlog"] = self.check_liquidation_backlog(liq_system) | |
| if trading_engine: | |
| ctx["orderbook_spread"] = self.check_orderbook_spread(trading_engine) | |
| # Log all results | |
| for check_name, result in ctx.items(): | |
| status = result.get("status", AuditStatus.SKIPPED) | |
| self.audit.log( | |
| category=AuditCategory.INTEGRITY, | |
| severity=AuditSeverity.HIGH if status == AuditStatus.FAILED else AuditSeverity.INFO, | |
| status=status, | |
| message=result.get("message", f"{check_name} check completed"), | |
| details=result.get("details", {}), | |
| actor="airmicrodrip_auditor", | |
| component=check_name, | |
| ) | |
| # Run base framework checks too | |
| base_report = self.audit.run_audit(context=ctx) | |
| return { | |
| "base_report_id": base_report.report_id, | |
| "overall_score": base_report.overall_score, | |
| "airmicrodrip_checks": ctx, | |
| "system_health": self.audit.get_system_health(), | |
| } | |
| if __name__ == "__main__": | |
| # Standalone demo | |
| auditor = AirMicroDripAuditor() | |
| print("AirMicroDrip Auditor initialized with checks:") | |
| for cid, check in auditor.audit.checks.items(): | |
| print(f" {cid}: {check.name} ({check.category.value}, {check.severity.value})") | |
| print(f"\nTotal checks registered: {len(auditor.audit.checks)}") | |