Data Flow

Relevant source files

The following files were used as context for generating this wiki page:

• ARCHITECTURE.md
• apps/data-processing/processors/forum_pipeline.py
• apps/backend/src/signals/signals.service.ts
• apps/backend/src/gateways/signal.gateway.ts
• apps/data-processing/README.md
• README.md

The OnChain Sage system is built upon a data-first architecture, emphasizing the efficient flow and processing of data as its core foundation. This modular, event-driven approach ensures that data from various sources is ingested, processed, and delivered in real-time to provide actionable trading insights. The entire system is designed for scalability and loose coupling between components, from initial data collection to final user interaction on the dashboard.

The data flow can be broken down into five key layers: Data Ingestion, Data Processing, Storage, Signal Delivery, and User Interaction. Each layer has distinct responsibilities, working in concert to transform raw data into valuable signals for crypto traders.

Sources: ARCHITECTURE.md

High-Level Data Flow

The following diagram illustrates the overall data flow, from external sources through the system's layers to the end-user.

This diagram shows data moving from external APIs into the ingestion and processing layers, where signals are generated. These signals are then stored and delivered to the frontend via REST APIs and WebSockets.

Sources: ARCHITECTURE.md, apps/data-processing/README.md

System Layers and Components

The data flow is managed across several distinct layers, each with specific components and responsibilities.

Layer	Technologies	Responsibilities	Source Files
Data Ingestion	Python, API Clients	Collects real-time social sentiment from X and on-chain market data from Raydium/Dex Screener.	ARCHITECTURE.md, apps/data-processing/README.md
Data Processing	Python, TensorFlow, PyTorch	Performs sentiment analysis, extracts metrics, generates trading signals, and aggregates data.	ARCHITECTURE.md, apps/data-processing/processors/forum_pipeline.py
Storage	PostgreSQL, Redis	Provides data persistence for signals and other entities, and caching for performance.	ARCHITECTURE.md, apps/backend/README.md
Signal Delivery	NestJS, WebSocket	Delivers processed signals to the frontend via REST APIs and real-time WebSocket updates.	ARCHITECTURE.md, apps/backend/src/gateways/signal.gateway.ts
User Interaction	Next.js, React	Visualizes data on a user-friendly dashboard and handles user interactions.	ARCHITECTURE.md, apps/frontend/README.md

Sources: ARCHITECTURE.md, apps/data-processing/README.md, apps/data-processing/processors/forum_pipeline.py, apps/backend/src/gateways/signal.gateway.ts

Detailed Data Flow Stages

1. Data Ingestion

The process begins with data collection from external sources. The data-processing service contains clients responsible for this task.

• X (Twitter) Scraper: Collects social media data and crypto-related tweets for sentiment analysis.
• Raydium Client: Interacts with the Raydium protocol to fetch on-chain liquidity and trading data.
• Other API Clients: Integrate with sources like Dex Screener for additional market data.

Sources: apps/data-processing/README.md, ARCHITECTURE.md

2. Data Processing

Once ingested, the raw data is processed to generate trading signals. The forum_pipeline.py provides a concrete example of how a trading call from a forum is processed.

The key steps in this pipeline are:

1. Parsing: A raw string (e.g., 'Buy ETH at $3000') is parsed into structured data (action, asset, price).
2. Data Enrichment: The current price of the asset is fetched.
3. Metric Calculation: Profitability is computed based on the entry price and current price.
4. Sentiment Analysis: A sentiment score is obtained for the asset.
5. Signal Formatting: The processed data is structured into a signal_data dictionary.
6. System Dispatch: The final signal is sent to other systems for storage and delivery.

The following diagram shows the flow within the process_forum_call function.

This flow demonstrates how raw text is transformed into a structured, enriched signal ready for distribution.

Sources: apps/data-processing/processors/forum_pipeline.py:78-112

Processed Signal Structure

The forum_pipeline.py prepares a dictionary with the following structure before sending it to downstream systems.

Field	Type	Description	Source
asset	string	The crypto asset ticker (e.g., 'ETH').	forum_pipeline.py:96
action	string	The trading action ('Buy' or 'Sell').	forum_pipeline.py:97
entry_price	float	The price at which the trade was called.	forum_pipeline.py:98
current_price	float	The current market price of the asset.	forum_pipeline.py:99
profitability	float	The calculated profit/loss percentage.	forum_pipeline.py:100
confidence	float	The sentiment score for the asset.	forum_pipeline.py:101
timestamp	string	The UTC timestamp in ISO format.	forum_pipeline.py:102

Sources: apps/data-processing/processors/forum_pipeline.py:95-103

3. Signal Delivery (Real-time)

After a signal is processed and stored, it needs to be delivered to connected clients in real-time. This is handled by the backend's WebSocket gateway. The SignalsService and SignalGateway work together to broadcast new signals.

1. The SignalsService contains a method broadcastNewSignal which is called when a new signal is created or updated.
2. This service method emits a 'newSignal' event using the server instance of the WebSocket gateway.
3. The SignalGateway listens for client connections and disconnections but primarily acts as the server endpoint for the SignalsService to broadcast messages.

The sequence diagram below illustrates this real-time update flow.

This shows how a signal generated by the data processing layer is pushed through the backend service to the WebSocket gateway, which then broadcasts it to all connected frontend clients.

Sources: apps/backend/src/signals/signals.service.ts:133-138, apps/backend/src/gateways/signal.gateway.ts

Conclusion

The data flow in OnChain Sage is a well-defined, multi-layered process designed for real-time performance and scalability. It starts with ingesting raw data from diverse on-chain and off-chain sources, moves through a sophisticated processing pipeline to generate and enrich trading signals, and concludes by delivering these signals instantly to users via WebSockets. This end-to-end flow is critical to the platform's ability to provide timely and actionable crypto trading insights.

Sources: ARCHITECTURE.md