gao/ljwx-stock

Fork 0

Files

T

bruno 95ee673378 phase1 completed

2025-11-10 23:27:16 +01:00

19 KiB

Raw Permalink Blame History

Stock Recommendation System - Data Pipeline Architecture

Executive Summary

The system implements a comprehensive data pipeline that fetches stock data from multiple sources, engineers features for ML models, generates predictions, and caches results for performance optimization. The pipeline integrates multiple data providers (TuShare, yfinance, OpenBB) with fallback mechanisms to ensure reliability.

1. Data Ingestion Layer

1.1 Data Sources and Collection Methods

Primary Data Sources:

TuShare API (Chinese A-stocks)
- Symbols: 6XXXXX (Shanghai), 0XXXXX (Shenzhen), 3XXXXX (Growth Enterprise Market)
- Data: OHLCV + fundamental metrics (PE, PB, market cap)
- Authentication: Token-based API key
- Default symbols: 600000, 600031, 601988, 601398
yfinance (Global stocks, fallback for Chinese stocks)
- Symbols: yfinance format (600000.SS, 000001.SZ)
- Data: OHLCV data, real-time quotes
- No authentication required
OpenBB SDK (Unified data provider - optional)
- Provides unified interface across multiple sources
- Token-based authentication
- Enhanced technical indicators and fundamental data

Data Fetcher Module (`/root/work/stock/backend/app/modules/data_fetcher.py`)

Key Classes:

DataFetcher: Main class handling data collection
Methods:
- fetch_stock_data(): Fetch OHLCV data for symbols with period/interval
- get_realtime_quote(): Real-time price quotes with caching
- fetch_fundamentals(): PE, PB, market cap, industry classification
- fetch_sector_data(): Sector-level performance metrics
- get_market_sentiment(): Overall market sentiment indicators

Fetching Strategy:

DataFetcher.fetch_stock_data(symbols, period='1y', interval='1d')
│
├─ For Chinese A-stocks (if TuShare token available):
│  ├─ Convert symbol format (600000 -> 600000.SH)
│  ├─ Call TuShare API with date range
│  └─ Normalize OHLCV columns
│
└─ Fallback to yfinance:
   ├─ Convert symbol format (600000 -> 600000.SS)
   ├─ Download with period and interval
   └─ Return DataFrame with [Open, High, Low, Close, Volume]

Data Output Format:

{
    'symbol': {
        'Open': float,
        'High': float,
        'Low': float,
        'Close': float,
        'Volume': int,
        'index': DatetimeIndex
    }
}

1.2 Realtime Quote API

Method: get_realtime_quote(symbols: List[str])

Features:

Uses yfinance 1-minute interval data for latest prices
Caches results with TTL: 60 seconds
Computes simulated bid/ask spread (±1% from Close)
Returns: {symbol: {price, bid, ask, volume, timestamp}}

Cache Key: stock:quote:{symbol}

1.3 Fundamental Data Fetching

Method: fetch_fundamentals(symbol: str)

Process:

Check cache (TTL: 1 hour)
Call TuShare API if token available
Fallback to default/empty values

Returns:

{
    'symbol': str,
    'name': str,
    'pe_ratio': float,
    'pb_ratio': float,
    'market_cap': float,
    'industry': str,
    'roe': float
}

Cache Key: stock:fundamentals:{symbol}

2. Data Transformation & Feature Engineering Layer

2.1 Feature Engineering Pipeline

Module: /root/work/stock/backend/app/modules/feature_engineer.py

Main Class: FeatureEngineer

Entry Point: engineer_features(df: pd.DataFrame) -> pd.DataFrame

Input Requirements:

DataFrame with columns: [Open, High, Low, Close, Volume]
Datetime index

Processing Flow:

Raw OHLCV Data
│
├─ Type Conversion (float/int normalization)
│
├─ Technical Features (if enabled):
│  ├─ Moving Averages
│  ├─ Momentum Indicators
│  ├─ Volatility Indicators
│  └─ Trend Indicators
│
└─ Output: Enhanced DataFrame with all features

2.2 Feature Categories and Calculations

A. Moving Average Features (5 features each period)

Periods: 5, 10, 20, 50, 200 days
MA{period}: Simple moving average
Price_MA{period}_Ratio: Close / MA (normalized price relative to trend)
MA5_MA20_Cross: Binary signal (1 if MA5 > MA20, else 0)
Total: 10 features + 1 signal

B. Momentum Indicators (8 features)

RSI_14: Relative Strength Index (14-period)
- Calculation: 100 - (100 / (1 + RS)) where RS = avg_gains / avg_losses
MACD: Moving Average Convergence Divergence
- MACD = EMA(12) - EMA(26)
- Signal = EMA(MACD, 9)
- MACD_Hist = MACD - Signal
Stochastic Oscillator:
- K = 100 * (Close - Low14min) / (High14max - Low14min)
- D = SMA(K, 3)
ROC_12: Rate of Change (12-period)
- ROC = pct_change(12) = (Close[t] - Close[t-12]) / Close[t-12]
Momentum_10: Price momentum (10-period)
- Momentum = Close[t] - Close[t-10]

C. Volatility Indicators (8 features)

ATR_14: Average True Range
- TR = max(High-Low, |High-Close[t-1]|, |Low-Close[t-1]|)
- ATR = SMA(TR, 14)
Bollinger Bands (4 features):
- BB_Middle = SMA(Close, 20)
- BB_Upper = Middle + 2*StdDev
- BB_Lower = Middle - 2*StdDev
- BB_Width = Upper - Lower
- BB_Position = (Close - Lower) / (Upper - Lower)
Historical Volatility:
- Volatility_20 = StdDev(pct_change(Close), 20)
- Volatility_60 = StdDev(pct_change(Close), 60)

D. Trend Indicators (4 features)

Momentum_5d: (Close[t] - Close[t-5]) / Close[t-5]
Momentum_20d: (Close[t] - Close[t-20]) / Close[t-20]
Volume_Ratio: Volume / SMA(Volume, 20)
Avg_Return_10d: SMA(pct_change(Close), 10)

Total Features Generated: 33+ technical features

2.3 Feature Normalization

Method: normalize_features(df, scaler=None, fit=False)

Process:

Separate feature columns from OHLCV
Use StandardScaler for Z-score normalization
Transform: (x - mean) / std_dev
Two modes:
- fit=True: Fit scaler on training data
- fit=False: Apply pre-fitted scaler (inference)

Output: Normalized DataFrame + scaler object for later use

3. ML Model Training and Prediction Layer

3.1 ML Training Module

Module: /root/work/stock/backend/app/modules/ml_trainer.py

Class: MLTrainer

Supported Models:

XGBoost (default)
- Objective: binary:logistic (classification)
- Max depth: 5
- Learning rate: 0.1
- Estimators: 100
LightGBM (alternative)
- Objective: binary classification
- Num leaves: 31
- Learning rate: 0.05
- Estimators: 100

3.2 Training Flow

Feature DataFrame (33+ features)
│
├─ Data Preparation:
│  ├─ Select feature columns
│  ├─ Fill NaN values
│  └─ Convert to numpy array
│
├─ Model Training:
│  ├─ Initialize XGBoost/LightGBM
│  ├─ Train on X_train, y_train
│  └─ Store model object
│
└─ Output: Trained classifier

3.3 Prediction and Scoring

Method: predict_scores(X: np.ndarray, normalize: bool=True)

Process:

Get probability predictions using predict_proba()
Extract probability of class 1
Normalize to 0-100 scale (multiply by 100)
Returns: Array of stock scores [0-100]

Score Interpretation:

0-30: Weak signal (Low recommendation score)
30-70: Neutral zone
70-100: Strong signal (High recommendation score)

3.4 Model Persistence

Methods:

save_model(path): Serialize with joblib to disk
load_model(path): Deserialize from disk
Default path: /app/models/stock_model.pkl

3.5 Dummy Model for Demo

Function: create_dummy_model()

Purpose: Generate synthetic model for testing/demo when real training not available

Process:

Generates 1000 samples with 33 features
Creates synthetic labels with correlation to features
Trains XGBoost classifier
Used when model not pre-trained

4. Data Caching and Storage Layer

4.1 Redis Cache Manager

Module: /root/work/stock/backend/app/modules/cache_manager.py

Class: CacheManager

Backend: Redis key-value store

Configuration:

Connection: redis://localhost:6379/0 (default)
Serialization: JSON for dict/list, string for primitives

4.2 Cache Keys and TTLs

Key Pattern	TTL	Purpose	Content
`stock:quote:{symbol}`	60s	Realtime prices	{price, bid, ask, volume, timestamp}
`stock:features:{symbol}`	300s	Engineered features	DataFrame features dict
`stock:score:{symbol}`	60s	ML prediction score	Float [0-100]
`stock:fundamentals:{symbol}`	3600s	Company metrics	{pe, pb, market_cap, industry}
`recommendation:top10:latest`	60s	Top 10 recommendations	List of recommendation dicts
`llm:explanation:{symbol}`	3600s	Stock analysis text	String (markdown)
`market:context:latest`	60s	Market sentiment	{sentiment, strength, volatility}

4.3 Cache Operations

Methods:

set(key, value, ttl): Store with expiration
get(key): Retrieve with JSON parsing
delete(key): Remove key
clear_prefix(prefix): Bulk delete with pattern
exists(key): Check existence
health_check(): Test Redis connection

5. Data Flow: Request to Recommendation

5.1 Full Pipeline Flow

User Request: GET /api/recommendations
│
├─ Check Cache:
│  └─ Cache hit → Return cached result
│
├─ Data Fetching Phase:
│  ├─ Get symbols from config [600000, 600031, 601988, 601398]
│  ├─ For each symbol:
│  │  ├─ Try TuShare (Chinese A-stocks)
│  │  └─ Fallback to yfinance
│  └─ Aggregate data_dict: {symbol: OHLCV_DataFrame}
│
├─ Feature Engineering Phase:
│  ├─ For each symbol:
│  │  ├─ engineer_features(df) → df_features (33+ features)
│  │  ├─ Get latest features df_features.iloc[-1:]
│  │  └─ Extract feature columns only
│  └─ Prepare X: [n_samples, n_features]
│
├─ ML Prediction Phase:
│  ├─ For each stock:
│  │  ├─ predict_scores(X) → score [0-100]
│  │  ├─ get_realtime_quote() → current_price, bid/ask
│  │  └─ fetch_fundamentals() → pe_ratio, pb_ratio, industry
│  └─ Aggregate scores dict
│
├─ Ranking Phase:
│  ├─ select_top_recommendations(scores, metadata, n=10)
│  └─ Sort by score descending
│
├─ LLM Explanation Phase:
│  ├─ Format recommendation prompt
│  ├─ Call Claude API via OpenRouter
│  └─ Get natural language explanation
│
├─ Cache Result:
│  └─ cache.set(recommendation_key, result, ttl=60)
│
└─ Return Response:
   └─ {timestamp, recommendations, market_context, explanation}

5.2 Single Stock Detail Request

User Request: GET /api/recommendations/{symbol}
│
├─ Fetch 1-year OHLCV data
│  ├─ TuShare → yfinance fallback
│  └─ Generate mock data if unavailable
│
├─ Engineer Features
│  └─ feature_engineer.engineer_features(df)
│
├─ Get Fundamentals & Quote
│  ├─ fetch_fundamentals(symbol)
│  └─ get_realtime_quote([symbol])
│
├─ Generate Analysis
│  ├─ Format stock analysis prompt
│  ├─ Call Claude LLM
│  └─ Get interpretation
│
├─ Prepare Price Chart
│  ├─ Last 30 days: dates, prices, volumes
│  └─ 20-day MA overlay
│
└─ Return:
   └─ {symbol, name, detail, analysis, price_chart}

6. Current Data Sources and Availability

6.1 Available Stock Symbols

Default Watchlist (China A-stocks):

600000  - Shanghai Pudong Development Bank
600031  - Shenergy Company Limited
601988  - China Bank
601398  - Industrial and Commercial Bank of China

Configuration Location: app/config.py

stock_symbols: list = ["600000", "600031", "601988", "601398"]

6.2 Data Availability at Each Stage

Stage 1: Data Ingestion

Format: Raw OHLCV DataFrame
Columns: Open, High, Low, Close, Volume
Index: DatetimeIndex (trading dates)
Length: Configurable lookback (default: 365 days)
Frequency: Daily candlesticks
Sample Size: ~252 trading days/year

Stage 2: Feature Engineering

Format: Enhanced DataFrame
Columns: Original 5 (OHLCV) + 33 technical features
Total Rows: Same as input (252 trading days)
NaN Handling: Forward/backward filled based on rolling calculations
Memory: ~50-100KB per symbol

Stage 3: ML Features

Format: 2D numpy array [1, 33]
Data: Latest features only (last row)
Normalization: StandardScaler applied
Type: float32
Shape: [1 sample, 33 features]

Stage 4: Predictions

Format: Float array
Range: 0.0-1.0 (before scaling) or 0-100 (after scaling)
Per Symbol: Single score value
Cache TTL: 60 seconds

Stage 5: Final Recommendations

Format: List of Dict
Per Item: {symbol, score, name, current_price, industry, pe_ratio, pb_ratio, predicted_return, confidence, technical_signals}
Ordering: Descending by score
Top N: 10 stocks
Cache TTL: 60 seconds

6.3 Fundamental Data

Available Metrics (from TuShare):

PE Ratio (市盈率)
PB Ratio (市净率)
Market Cap (总市值)
Industry Classification (行业分类)
Stock Name (股票名称)
Area (地域)
List Date (上市日期)

Optional Metrics (if TuShare unavailable):

Default values set to 0.0
Name extracted from symbol

6.4 Market-Level Data

Sector Performance:

10 sectors with performance scores [0-1]:
- 金融 (Finance): 0.5
- 房地产 (Real Estate): 0.4
- 医药生物 (Pharma/Bio): 0.6
- 消费 (Consumer): 0.55
- 工业 (Industrial): 0.45
- 技术 (Technology): 0.7
- 能源 (Energy): 0.35
- 材料 (Materials): 0.4
- 电信 (Telecom): 0.45
- 公用事业 (Utilities): 0.3

Market Sentiment:

{
    "overall": "bullish|neutral|bearish",
    "strength": float [0-1],
    "volatility": float [0-1],
    "volume": "high|medium|low",
    "trend": "uptrend|downtrend|ranging"
}

7. Database and Data Persistence

7.1 Database Configuration

Module: /root/work/stock/backend/app/database.py

Type: PostgreSQL (production) or configured via DATABASE_URL

Default Connection:

postgresql://stock_user:stock_pass@localhost:5432/stock_db

Features:

Connection pooling (10 connections, 20 overflow)
Connection timeout: 10s
Statement timeout: 30s
Health check: ping on connect

7.2 Data Models

ORM Base: SQLAlchemy declarative base

Tables (Planned but not detailed in code):

Stock metadata
Historical quotes
Recommendations history
Backtest results
User portfolios

8. Data Quality and Validation

8.1 Data Validation

Input Validation:

Required columns check: [Open, High, Low, Close, Volume]
Type conversion to float/int
NaN removal after feature engineering
Volume/Price > 0

Feature Validation:

Standard deviation > 0 for volatility calculations
Epsilon (1e-6) added to prevent division by zero
Min periods for rolling calculations

8.2 Fallback Mechanisms

Multi-source Fallback Strategy:

TuShare (preferred for Chinese)
    └─→ yfinance (universal fallback)
        └─→ Mock data (demo mode)

Fundamentals:
TuShare → yfinance → Default values

8.3 Error Handling

Try-Except Pattern:

Graceful degradation on API failures
Continues processing other symbols on individual errors
Mock data generation for demo purposes
Detailed error logging

9. Performance Characteristics

9.1 Data Volume

Per Stock:

1-year data: ~252 trading days
Raw OHLCV: 5 columns × 252 rows = 1,260 values
With features: 38 columns × 252 rows = 9,576 values
Uncompressed size: ~40-50 KB

For 4 stocks:

Total: 160-200 KB in memory
Cached fundamentals: ~1 KB per stock
ML features: ~1.3 KB per stock

9.2 Computation Time

Estimated Latencies:

Data fetch: 1-3 seconds per symbol
Feature engineering: 50-200ms per symbol
ML prediction: 10-50ms per symbol
Total pipeline: 5-15 seconds for 4 stocks
Cached result return: <10ms

9.3 Cache Efficiency

Hit Rates:

Quotes: High (60s TTL, frequent requests)
Features: Medium (300s TTL)
Recommendations: High (60s TTL, popular endpoint)
Fundamentals: High (1 hour TTL)

10. Configuration and Environment Variables

10.1 Data Pipeline Configuration

File: /root/work/stock/backend/app/config.py

Variable	Default	Purpose
`TUSHARE_TOKEN`	(required)	API authentication for TuShare
`OPENBB_TOKEN`	(optional)	OpenBB SDK authentication
`STOCK_SYMBOLS`	"600000,600031,601988,601398"	Default watchlist
`LOOKBACK_DAYS`	365	Historical data range
`REBALANCE_INTERVAL`	"daily"	Portfolio rebalancing frequency
`ENABLE_TECHNICAL_FEATURES`	true	Enable feature engineering
`ENABLE_FUNDAMENTAL_FEATURES`	true	Include fundamental metrics
`ENABLE_MOMENTUM_FEATURES`	true	Include momentum indicators
`MIN_CONFIDENCE`	0.5	Score threshold for recommendations
`CACHE_TTL_QUOTE`	60	Quote cache duration (seconds)
`CACHE_TTL_FEATURES`	300	Feature cache duration
`REDIS_URL`	"redis://localhost:6379/0"	Cache backend

11. Data Flow Diagram

INGESTION → TRANSFORMATION → ML MODEL → RANKING → OUTPUT
    ↓            ↓              ↓          ↓         ↓
  TuShare    Features (33)   Prediction  Top10   Recommendation
  yfinance   Normalized      Score [0]   Select   + Explanation
 (Fallback)   Features        with       by       + Charts
              Caching         Cache      Score    + Cache
              [300s TTL]     [60s TTL]  [60s TTL]

EXTERNAL APIs:
├─ TuShare API → Real fundamentals (PE, PB, etc.)
├─ yfinance → Global OHLCV data
├─ OpenBB SDK → Unified data (optional)
└─ OpenRouter Claude → LLM explanations

STORAGE:
├─ Redis → Live data cache
├─ PostgreSQL → Historical data, recommendations, backtest results
└─ Filesystem → ML models (/app/models/stock_model.pkl)

12. Data Integration Points

12.1 API Endpoints Using Data Pipeline

Endpoint	Data Sources	Cache Used	Output Format
`/api/recommendations`	TuShare/yfinance	recommendation	JSON list
`/api/recommendations/{symbol}`	Full 1-year data	features, analysis	JSON detail
`/api/analysis/sectors`	Hardcoded sector data	-	JSON dict
`/api/analysis/signals/{symbol}`	Simulated signals	-	JSON
`/api/analysis/market`	Market sentiment	-	JSON
`/api/backtest/run`	Historical recommendations	-	JSON metrics

12.2 Data Refresh Strategy

Recommendations: Every 60 seconds (cache TTL) Quotes: Every 60 seconds Features: Every 300 seconds Fundamentals: Every 1 hour Full Pipeline: Triggered on user request

Summary Table: Data Availability

Component	Data Type	Source	Format	Update Frequency
Raw OHLCV	Time series	TuShare/yfinance	DataFrame	On request
Technical Features	Indicators	Feature Engineer	33+ columns	On request
ML Scores	Predictions	XGBoost/LightGBM	0-100	Per request
Fundamentals	Company metrics	TuShare	Dict	1 hour cache
Realtime Quotes	Price data	yfinance	Dict	60s cache
Sector Data	Market data	Static config	Dict	On request
Market Sentiment	Indicators	Simulated	Dict	On request
Recommendations	Top N stocks	ML Ranking	JSON list	60s cache
LLM Analysis	Text explanation	OpenRouter Claude	String	1 hour cache

19 KiB Raw Permalink Blame History Unescape Escape

Stock Recommendation System - Data Pipeline Architecture

Executive Summary

1. Data Ingestion Layer

1.1 Data Sources and Collection Methods

Primary Data Sources:

Data Fetcher Module (/root/work/stock/backend/app/modules/data_fetcher.py)

1.2 Realtime Quote API

1.3 Fundamental Data Fetching

2. Data Transformation & Feature Engineering Layer

2.1 Feature Engineering Pipeline

2.2 Feature Categories and Calculations

A. Moving Average Features (5 features each period)

B. Momentum Indicators (8 features)

C. Volatility Indicators (8 features)

D. Trend Indicators (4 features)

2.3 Feature Normalization

3. ML Model Training and Prediction Layer

3.1 ML Training Module

3.2 Training Flow

3.3 Prediction and Scoring

3.4 Model Persistence

3.5 Dummy Model for Demo

4. Data Caching and Storage Layer

4.1 Redis Cache Manager

4.2 Cache Keys and TTLs

4.3 Cache Operations

5. Data Flow: Request to Recommendation

5.1 Full Pipeline Flow

5.2 Single Stock Detail Request

6. Current Data Sources and Availability

6.1 Available Stock Symbols

6.2 Data Availability at Each Stage

Stage 1: Data Ingestion

Stage 2: Feature Engineering

Stage 3: ML Features

Stage 4: Predictions

Stage 5: Final Recommendations

6.3 Fundamental Data

6.4 Market-Level Data

7. Database and Data Persistence

7.1 Database Configuration

7.2 Data Models

8. Data Quality and Validation

8.1 Data Validation

8.2 Fallback Mechanisms

8.3 Error Handling

9. Performance Characteristics

9.1 Data Volume

9.2 Computation Time

9.3 Cache Efficiency

10. Configuration and Environment Variables

10.1 Data Pipeline Configuration

11. Data Flow Diagram

12. Data Integration Points

12.1 API Endpoints Using Data Pipeline

12.2 Data Refresh Strategy

Summary Table: Data Availability

19 KiB

Raw Permalink Blame History

Data Fetcher Module (`/root/work/stock/backend/app/modules/data_fetcher.py`)