Skip to content

Public security notice: This documentation is intentionally redacted. Sensitive server paths, private keys, secret tokens, and origin network details are removed.

Core Engine & Workflow

This page describes how the engine orchestrates a full DeFi token risk assessment, from command-line entrypoint to final reports.

The reference implementation lives primarily in:

  • scripts/engine/defi_complete_risk_assessment_clean.py

Dashboard helpers plug into the same data directories and report outputs.

High-level flow

At a high level, a full run does the following:

  1. Initializes environment and paths
  2. Determines PROJECT_ROOT, DATA_DIR, LOGS_DIR.
  3. Ensures these directories exist.

  4. Sets timestamped paths such as:

    • RISK_REPORT_DIR
    • EXCEL_REPORT_PATH, RISK_REPORT_JSON, RISK_REPORT_CSV
    • *_LATEST aliases for dashboards.

  5. Initializes shared services

  6. Optional cache_manager from cache_manager.get_cache_manager(DATA_DIR)
  7. Optional error_handler from api_error_handler.get_error_handler(DATA_DIR)
  8. SharedAPIScheduler with a global worker pool (per-API caps live in APIErrorHandler)

  9. HTTP request state and rate-limit tracking structures.

  10. Loads configuration and datasets

  11. Token universe from DATA_DIR/tokens.csv.
  12. Fallback definitions from DATA_DIR/fallbacks/fallbacks.json.
  13. Token Data Viewer snapshot from DATA_DIR/token_data_viewer.csv.
  14. EU-mode configuration and regulated stablecoin allow-list.

  15. Caching and request policy from DATA_DIR/settings.json.

  16. Processes tokens in batches

  17. For each token (address + symbol + chain), the engine:

    1. Builds a per-token context (chain, symbol, tags, maybe allow-list status).
    2. Submits many data-fetch tasks through the shared scheduler.
    3. Applies caching and rate-limit-aware HTTP logic.
    4. Normalizes responses into an internal token_entry structure.

  18. Calculates scores and red flags

  19. Derives component quality scores (1–10) for behavioral categories.
  20. Converts them into risk contributions (0–10) using weights.
  21. Applies red-flag boosts and market-structure penalties.

  22. Aggregates into a final risk score and credibility signals.

  23. Persists reports

  24. Writes JSON, CSV and XLSX reports to timestamped files.
  25. Updates *_latest aliases for dashboards and downstream tooling.
  26. Optionally generates a social score analysis report.

SharedAPIScheduler & concurrency model

The engine avoids per-token thread pools and instead uses a shared scheduler so that concurrency limits are enforced globally.

Key object, SharedAPIScheduler in scripts/engine/defi_complete_risk_assessment_clean.py (simplified):

```python class SharedAPIScheduler: def init(self, max_workers: int = 12, api_caps: Optional[Dict[str, int]] = None) -> None: self.max_workers = max(1, int(max_workers)) self.executor = ThreadPoolExecutor(max_workers=self.max_workers) self.api_caps: Dict[str, int] = dict(api_caps or {}) self._semaphores = { str(api).lower(): threading.BoundedSemaphore(max(1, int(cap))) for api, cap in self.api_caps.items() }

def submit(self, api_name: str, fn: Callable[..., Any], *args, **kwargs):
    ...
    return self.executor.submit(wrapped)

```

Usage pattern:

  • Heavy data-fetch tasks call get_shared_api_scheduler().submit("service_name", fn, ...).
  • The scheduler enforces a global thread limit (max_workers). It can also hold per-service semaphores in _semaphores, but get_shared_api_scheduler() builds the shared instance with an empty API_CONCURRENCY_CAPS: per-API concurrency caps are enforced centrally by APIErrorHandler in scripts/engine/api_error_handler.py instead.

Why this matters

  • Prevents nested executors from oversubscribing CPU/network.
  • Keeps per-API concurrency caps in one place, so providers that throttle aggressively stay capped even when many tokens are processed at once.
  • Keeps CPU and I/O usage more predictable in production environments.

Token processing lifecycle

Conceptually, each token goes through the following stages:

  1. Discovery: read from tokens.csv (address, symbol, chain, labels).
  2. Bootstrap: attach cached viewer/fallback data if available.
  3. Data fetch: schedule multiple API calls through the scheduler.
  4. Normalization: consolidate raw responses into a uniform internal model.
  5. Scoring: compute behavioral component scores and overall risk.
  6. Persistence: write to JSON/CSV/XLSX and, optionally, social reports.

The implementation is structured so that the scoring pipeline can be reused with different data sources (e.g. live-only vs. cache-heavy modes) without changing the core model.

Cache manager and error handler integration

Two plug-in style components enhance the core script:

  • cache_manager – centralizes cache operations and policies.
  • error_handler – manages per-service health, cooldowns and error logging.

When available, cache_manager is preferred over the legacy JSON cache helpers. The engine module initializes it once the data and log directories exist:

python try: from cache_manager import get_cache_manager cache_manager = get_cache_manager(DATA_DIR) print("✅ Enhanced cache manager initialized") except ImportError: print("⚠️ Enhanced cache manager not available, using basic cache") cache_manager = None

Similarly, the enhanced API error handler is loaded right after it, if present:

python try: from api_error_handler import get_error_handler error_handler = get_error_handler(DATA_DIR) print("✅ Enhanced API error handler initialized") except ImportError: print("⚠️ Enhanced API error handler not available, using basic error handling") error_handler = None

If these modules are not importable, the engine falls back to embedded basic cache and error-handling logic.

Where to go next