Data Processing Pipeline
Relevant source files
The following files were used as context for generating this wiki page:
The Data Processing Pipeline is a core component of the OnChainSage project, responsible for ingesting, processing, and analyzing data from various sources to generate actionable trading signals. It operates as a modular, event-driven system that combines social sentiment from platforms like X (Twitter) with on-chain market data from sources such as Raydium and Dex Screener.
This pipeline is designed to be data-first, ensuring that the flow and transformation of data drive the system's functionality. It handles everything from raw data collection to final storage in a persistent database and cache, making the insights available to the rest of the OnChainSage ecosystem.
Sources: ARCHITECTURE.md, apps/data-processing/README.md, apps/data-processing/processors/pipeline.py
Architecture and Core Components
The pipeline is orchestrated by the Pipeline class, which initializes and manages several specialized components. Each component has a distinct responsibility, promoting modularity and maintainability.
Sources: apps/data-processing/processors/pipeline.py:27-50
The following diagram illustrates the main components of the Pipeline class and their relationships.
Key Components
| Component | Description | Source File |
|---|---|---|
clients | A dictionary of client objects for interacting with external data sources like X, Raydium, Dex Screener, and forums. | apps/data-processing/processors/pipeline.py:42 |
DataValidator | Responsible for ensuring the quality and consistency of ingested data before processing. | apps/data-processing/processors/pipeline.py:43 |
CompositeSentimentAnalyzer | Analyzes the sentiment of social media data (e.g., tweets) using keyword-based analysis. | apps/data-processing/processors/pipeline.py:44 |
ProfitabilityCalculator | Calculates the profitability of trading calls from forums based on entry and current prices. | apps/data-processing/processors/pipeline.py:45 |
SignalAggregator | Combines sentiment data, market metrics, and profitability analysis to generate final trading signals. | apps/data-processing/processors/pipeline.py:46 |
Storage | Manages data persistence, handling storage in PostgreSQL and caching in Redis. | apps/data-processing/processors/pipeline.py:47 |
Sources: apps/data-processing/processors/pipeline.py:42-47, apps/data-processing/README.md
Configuration
The pipeline's behavior is configured via a YAML file, typically located at config/pipeline.yaml. The Pipeline class loads this configuration upon initialization to set up its components, monitoring services, and database connections.
Sources: apps/data-processing/processors/pipeline.py:34-36, apps/data-processing/processors/pipeline.py:261-262
| Configuration Key | Description |
|---|---|
monitoring | Contains settings for Sentry (DSN, environment) and Prometheus (port). |
sentiment | Configuration for the CompositeSentimentAnalyzer. |
signal | Configuration for the SignalAggregator. |
storage | Contains connection details for PostgreSQL and Redis. |
Sources: apps/data-processing/processors/pipeline.py:38-49
Execution Flow
The pipeline execution is managed by the run() method, which orchestrates the entire process in three main stages: Ingestion, Processing, and Storage. The entire run is timed using a Prometheus histogram.
# apps/data-processing/processors/pipeline.py:236-253
async def run(self):
"""Run the pipeline."""
try:
with PROCESSING_TIME.time():
# Ingest data
logger.info("Starting data ingestion")
data = await self.ingest_data()
logger.info("Data ingestion complete")
# Process data
logger.info("Starting data processing")
processed = await self.process_data(data)
logger.info("Data processing complete")
# Store results
logger.info("Starting data storage")
await self.store_data(processed)
logger.info("Data storage complete")
except Exception as e:
logger.error(f"Pipeline run failed: {str(e)}")
sentry_sdk.capture_exception(e)
raise
The following diagram shows the high-level execution flow of the pipeline.
1. Data Ingestion
The ingest_data method asynchronously collects data from all configured clients.
This diagram illustrates the sequence of data ingestion from various sources.
The collected data includes:
- Tweets: Fetched from the X API for the last 24 hours.
- Liquidity Pools: Fetched from the Raydium client.
- Token Metrics: Fetched from the Dex Screener client for each pool.
- Forum Calls: Fetched from the forum client for the last 7 days.
Sources: apps/data-processing/processors/pipeline.py:75-100
2. Data Processing
The process_data method takes the raw ingested data and transforms it into structured, analyzed information and trading signals.
This flowchart shows the steps involved in processing the ingested data.
The processing steps are:
- Validation: The
DataValidatorfilters the raw data to ensure only valid records are processed. - Sentiment Analysis: The
CompositeSentimentAnalyzerprocesses valid tweets to determine their sentiment score. - Profitability Calculation: The
ProfitabilityCalculatordetermines the performance of trading calls from forums. - Signal Generation: The
SignalAggregatorcombines sentiments, market metrics, and profitability data to generate the final trading signals.
Sources: apps/data-processing/processors/pipeline.py:122-155
3. Data Storage
The store_data method persists the processed data and generated signals.
- PostgreSQL: Valid tweets, market metrics, forum calls, and generated signals are stored in a PostgreSQL database for long-term persistence.
- Redis: The latest signals and market metrics are cached in Redis for fast access, with specified TTLs (Time To Live).
latest_signals: Cached for 1 hour.latest_market_metrics: Cached for 5 minutes.
Sources: apps/data-processing/processors/pipeline.py:180-213
Monitoring and Error Handling
The pipeline is integrated with monitoring and error-handling tools to ensure reliability.
Prometheus Metrics
Prometheus is used to collect and expose key performance metrics from the pipeline. The server is started on the port defined in the configuration.
| Metric | Type | Description |
|---|---|---|
processed_tweets_total | Counter | Number of tweets processed. |
processed_pools_total | Counter | Number of liquidity pools processed. |
processed_forum_calls_total | Counter | Number of forum calls processed. |
generated_signals_total | Counter | Number of trading signals generated. |
processing_time_seconds | Histogram | Time spent processing data in a pipeline run. |
signal_score | Gauge | The score of a generated trading signal, labeled by token and category. |
Sources: apps/data-processing/processors/pipeline.py:19-25, apps/data-processing/processors/pipeline.py:53-54
Sentry Error Tracking
Sentry is used for real-time error tracking. It is initialized with a DSN and environment from the configuration file. Any exceptions that occur during the ingest_data, process_data, or store_data stages are captured by Sentry, providing developers with detailed error reports.
Sources: apps/data-processing/processors/pipeline.py:38-41, apps/data-processing/processors/pipeline.py:104-105, apps/data-processing/processors/pipeline.py:171-172, apps/data-processing/processors/pipeline.py:217-218
Entry Point
The pipeline can be executed as a script. It reads the configuration path from the PIPELINE_CONFIG environment variable, with a default fallback to config/pipeline.yaml.
# apps/data-processing/processors/pipeline.py:265-268
if __name__ == '__main__':
# ...
config_path = os.getenv('PIPELINE_CONFIG', 'config/pipeline.yaml')
asyncio.run(Pipeline.create_and_run(config_path))
The setup.py file also defines a console script entry point named onchainsage, although its direct relation to running this specific pipeline is defined in onchainsage.cli:main, which is not provided.
Sources: apps/data-processing/processors/pipeline.py:256-268, apps/data-processing/setup.py:22-26