12 KiB
12 KiB
Data Pipeline - Quick Reference Guide
Pipeline Overview
┌─────────────────────────────────────────────────────────────────────┐
│ STOCK RECOMMENDATION SYSTEM DATA PIPELINE │
└─────────────────────────────────────────────────────────────────────┘
DATA INGESTION TRANSFORMATION ML LAYER RANKING
(Raw) (Features) (Prediction) (Top10)
TuShare Feature XGBoost Select
yfinance → Engineer → LightGBM → Top 10
(OHLCV) (33 Features) (0-100 Scores) (Scores)
↓
┌─────────────────────────────────────────────────────┬──────────────────────┐
│ CACHING LAYER (Redis) │ LLM EXPLANATION │
│ - Quotes (60s) - Features (300s) │ Claude API │
│ - Fundamentals (1h) - Recommendations (60s) │ (Narrative) │
└─────────────────────────────────────────────────────┴──────────────────────┘
↓
FINAL OUTPUT
(JSON Response to User)
1. Data Fetching Module
File: backend/app/modules/data_fetcher.py
Key Methods:
| Method | Purpose | Input | Output | Cache |
|---|---|---|---|---|
fetch_stock_data() |
OHLCV data | symbols, period='1y' | Dict[symbol: DataFrame] | N/A |
get_realtime_quote() |
Current prices | symbols | Dict[symbol: quote_data] | 60s |
fetch_fundamentals() |
PE, PB, market cap | symbol | Dict metrics | 1h |
fetch_sector_data() |
Sector performance | - | Dict[sector: score] | N/A |
get_market_sentiment() |
Market indicators | - | Dict sentiment | N/A |
Data Flow:
Symbol (e.g., "600000")
↓
[Check TuShare Token]
├─YES→ Convert to TS code (600000.SH)
│ ├─ Call TuShare API
│ └─ Normalize columns
└─NO→ Use yfinance
├─ Convert to yf format (600000.SS)
├─ Download data
└─ Return DataFrame
↓
Output: {Open, High, Low, Close, Volume}
Index: DatetimeIndex (trading dates)
2. Feature Engineering Module
File: backend/app/modules/feature_engineer.py
Generated Features (33 Total):
Moving Averages (11 features)
- MA5, MA10, MA20, MA50, MA200
- Price_MA5_Ratio, Price_MA10_Ratio, Price_MA20_Ratio, Price_MA50_Ratio, Price_MA200_Ratio
- MA5_MA20_Cross
Momentum Indicators (8 features)
- RSI_14 (Relative Strength Index)
- MACD, MACD_Signal, MACD_Hist
- Stoch_K, Stoch_D (Stochastic Oscillator)
- ROC_12 (Rate of Change)
- Momentum_10
Volatility Indicators (8 features)
- ATR_14 (Average True Range)
- BB_Middle, BB_Upper, BB_Lower, BB_Width (Bollinger Bands)
- BB_Position
- Volatility_20, Volatility_60
Trend Indicators (4 features)
- Momentum_5d, Momentum_20d
- Volume_Ratio
- Avg_Return_10d
Processing Pipeline:
Raw OHLCV DataFrame (5 cols × 252 rows)
↓
Type Conversion (float/int)
↓
[Technical Features Enabled?]
├─YES→ Generate 33 features
└─NO→ Skip feature engineering
↓
Feature DataFrame (38 cols × 252 rows)
↓
Normalize Features (StandardScaler)
├─ Mean-center each feature
└─ Scale by std deviation
↓
Output: Normalized features [0-100 scale]
Memory: ~50-100KB per symbol
3. ML Model Layer
File: backend/app/modules/ml_trainer.py
Model Architecture:
INPUT: Features (33 features)
↓
[XGBoost Classifier]
max_depth: 5
learning_rate: 0.1
n_estimators: 100
↓
OUTPUT: Binary probability [0-1]
↓
Multiply by 100
↓
SCORE: [0-100]
Prediction Flow:
Latest Features (1 sample × 33 features)
↓
model.predict_proba(X)
↓
Get probability of class 1
↓
Normalize: proba * 100
↓
Score: 0-100 (higher = stronger recommendation)
Score Interpretation:
0-30 ▯▯▯▯▯ Weak Signal (Sell/Avoid)
30-50 ▯▯▯▯▯ Neutral (Hold)
50-70 ▯▯▯▯▯ Positive Signal (Consider Buy)
70-100 ▯▯▯▯▯ Strong Signal (Buy)
4. Caching System
File: backend/app/modules/cache_manager.py
Cache Architecture:
┌─────────────────────────────────────┐
│ Redis Cache (in-memory) │
├─────────────────────────────────────┤
│ Key Pattern │ TTL │ Size │
├─────────────────────────────────────┤
│ stock:quote:* │ 60s │ 0.5KB │
│ stock:features:* │ 300s │ 50KB │
│ stock:score:* │ 60s │ 0.1KB │
│ stock:fundamentals:* │ 3600s │ 1KB │
│ recommendation:top10:latest │ 60s │ 5KB │
│ llm:explanation:* │ 3600s │ 10KB │
│ market:context:latest │ 60s │ 0.5KB │
└─────────────────────────────────────┘
Cache Hit Rates:
Quotes: ▰▰▰▰▰ 90% (60s TTL, frequent)
Features: ▰▰▰▯▯ 70% (300s TTL)
Fundamentals: ▰▰▰▰▯ 85% (1h TTL)
Recommendations: ▰▰▰▰▰ 95% (60s TTL)
5. Full Request Flow
GET /api/recommendations
REQUEST
↓
[Check Cache: recommendation:top10:latest]
├─HIT→ Return cached (< 10ms)
└─MISS→ Continue to pipeline
↓
FETCH DATA (1-3s per symbol)
├─ Symbol 1: 600000
├─ Symbol 2: 600031
├─ Symbol 3: 601988
└─ Symbol 4: 601398
↓
ENGINEER FEATURES (50-200ms each)
└─ 33 technical indicators per stock
↓
PREDICT SCORES (10-50ms each)
└─ ML model inference: 0-100
↓
GET METADATA (cached fundamentals)
├─ Current price
├─ PE/PB ratios
└─ Industry
↓
RANK & SELECT Top 10
└─ Sort by score descending
↓
GENERATE LLM EXPLANATION (optional)
├─ Format recommendation prompt
├─ Call Claude API (OpenRouter)
└─ Get narrative (1-3s)
↓
CACHE RESULT (60s TTL)
└─ Set: recommendation:top10:latest
↓
RESPONSE (JSON)
└─ {recommendations, market_context, explanation}
6. Data Availability by Component
Input Data (Stage 1: Ingestion)
Source: TuShare / yfinance
Format: DataFrame
├─ Open (float)
├─ High (float)
├─ Low (float)
├─ Close (float)
├─ Volume (int)
└─ Index (DatetimeIndex)
Period: 1 year (365 days)
Trading Days: ~252
Memory: 40-50 KB/stock
Transformed Data (Stage 2: Features)
Source: Feature Engineer
Format: DataFrame
├─ OHLCV (5 columns)
├─ Technical Features (33 columns)
└─ Index: DatetimeIndex
Total Columns: 38
Total Rows: ~252
Memory: 100-150 KB/stock
ML Features (Stage 3: Model Input)
Source: Latest row from Stage 2
Format: NumPy array
├─ Shape: [1, 33] (1 sample, 33 features)
├─ Dtype: float32
├─ Normalized: [-3 to +3] (Z-score)
└─ Memory: < 1 KB
Predictions (Stage 4: Scores)
Source: XGBoost/LightGBM
Format: Float array
├─ Value Range: 0-100
├─ Per Stock: 1 value
├─ 4 Stocks: 4 values
└─ Memory: < 1 KB
Cached TTL: 60s
Final Output (Stage 5: Recommendations)
Source: Ranking + LLM
Format: JSON
├─ timestamp
├─ recommendations (list)
│ ├─ symbol
│ ├─ score
│ ├─ name
│ ├─ current_price
│ ├─ industry
│ ├─ pe_ratio
│ ├─ pb_ratio
│ ├─ predicted_return
│ ├─ confidence
│ └─ technical_signals
├─ market_context
└─ explanation
Total Size: ~5-10 KB
Cached TTL: 60s
7. Configuration Keys
Essential Environment Variables:
# Data Sources
TUSHARE_TOKEN=<required for Chinese stocks>
OPENBB_TOKEN=<optional for unified data>
# Stock Selection
STOCK_SYMBOLS=600000,600031,601988,601398
LOOKBACK_DAYS=365
# Feature Engineering
ENABLE_TECHNICAL_FEATURES=true
ENABLE_FUNDAMENTAL_FEATURES=true
ENABLE_MOMENTUM_FEATURES=true
# ML Model
MODEL_TYPE=xgboost
MODEL_PATH=/app/models/stock_model.pkl
# Caching
REDIS_URL=redis://localhost:6379/0
CACHE_TTL_QUOTE=60
CACHE_TTL_FEATURES=300
CACHE_TTL_RECOMMENDATION=60
CACHE_TTL_LLM=3600
# LLM
OPENROUTER_KEY=<required for explanations>
LLM_MODEL=anthropic/claude-4.5-sonnet
LLM_TEMPERATURE=0.7
LLM_MAX_TOKENS=500
LLM_ENABLE_CACHE=true
8. Performance Metrics
Latency Breakdown (4 stocks, cache miss):
Data Fetching: 3-5s ▰▰▰▰▯
Feature Engineering: 0.5s ▰▯▯▯▯
ML Prediction: 0.3s ▰▯▯▯▯
Ranking & Metadata: 1s ▰▯▯▯▯
LLM Explanation: 2-3s ▰▰▯▯▯
─────────────────────────────────────
Total (no cache): 7-14s ▰▰▰▯▯
With cache: < 10ms ▯▯▯▯▯
Data Volume per Request:
Downloaded Data: 160-200 KB (OHLCV)
Engineered Features: 300-400 KB (all features)
ML Features Used: 0.1 KB (latest row)
API Response: 5-10 KB (JSON)
Cache Efficiency:
Typical Traffic Pattern:
Every 60s: New recommendations request (cache miss)
Remaining 59s: Quote/detail requests (cache hits)
Hit Rate: ~95% (most requests use cache)
9. Data Quality Checks
Input Validation:
✓ Required columns [Open, High, Low, Close, Volume]
✓ Type conversion to float/int
✓ Volume > 0
✓ Close > 0
✓ High >= Low
✓ Close between Low and High
Feature Validation:
✓ No NaN in final features
✓ Standard deviation > 0
✓ Epsilon (1e-6) prevents division by zero
✓ Rolling calculations use min_periods
Fallback Strategy:
TuShare API
↓ (if fails)
yfinance
↓ (if both fail)
Mock data for demo
10. Key Data Files
| File | Purpose | Location |
|---|---|---|
| data_fetcher.py | Data collection | backend/app/modules/ |
| feature_engineer.py | Feature generation | backend/app/modules/ |
| ml_trainer.py | Model training/prediction | backend/app/modules/ |
| cache_manager.py | Redis caching | backend/app/modules/ |
| config.py | Configuration | backend/app/ |
| database.py | Data persistence | backend/app/ |
| recommendations.py | API endpoint | backend/app/routes/ |
11. Troubleshooting Guide
No Data Returned:
Check:
1. TUSHARE_TOKEN valid? → View in .env
2. yfinance working? → Test with pip
3. Symbols correct? → 600000.SS vs 600000.SH
4. Network connection? → Ping API endpoints
Slow Performance:
Check:
1. Redis connection? → health_check()
2. Cache hit rate? → Monitor CACHE_KEYS
3. Feature engineering? → Slow rolling calculations?
4. ML model size? → Check MODEL_PATH
Feature NaN Values:
Causes:
- Insufficient lookback period (rolling window > data)
- Missing data in source (holidays, gaps)
- Invalid calculations (division by zero)
Fixes:
- Increase LOOKBACK_DAYS
- Fill NaN: fillna(method='ffill')
- Add epsilon to denominators
12. Performance Optimization Tips
Cache Optimization:
✓ Increase TTLs for stable data (fundamentals)
✓ Use cache.clear_prefix() for batch updates
✓ Monitor Redis memory usage
✓ Implement cache warming for top stocks
Feature Engineering:
✓ Disable unnecessary features (config)
✓ Use numpy for vectorized operations
✓ Cache features for 5min (medium TTL)
✓ Parallelize across stocks
ML Model:
✓ Reduce tree depth for faster inference
✓ Use LightGBM for faster training
✓ Cache model in memory
✓ Batch predictions when possible