gao/ljwx-stock

Fork 0

Files

T

bruno 95ee673378 phase1 completed

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

20 KiB

Raw Permalink Blame History

Data Pipeline - File Map and Code Locations

Directory Structure

/root/work/stock/
├── backend/
│   ├── main.py                          [FastAPI app entry point]
│   ├── app/
│   │   ├── __init__.py
│   │   ├── config.py                   [Settings & environment config]
│   │   ├── database.py                 [PostgreSQL connection]
│   │   ├── modules/                    [Core pipeline modules]
│   │   │   ├── __init__.py
│   │   │   ├── data_fetcher.py         [DATA INGESTION - TuShare/yfinance]
│   │   │   ├── feature_engineer.py     [TRANSFORMATION - 33 features]
│   │   │   ├── ml_trainer.py           [ML LAYER - XGBoost/LightGBM]
│   │   │   ├── cache_manager.py        [CACHING - Redis]
│   │   │   ├── llm_ranker.py           [LLM - Claude API explanations]
│   │   │   ├── backtester.py           [Strategy backtesting]
│   │   │   └── openbb_provider.py      [Alternative data provider]
│   │   ├── routes/                     [API endpoints]
│   │   │   ├── __init__.py
│   │   │   ├── recommendations.py      [/api/recommendations - uses full pipeline]
│   │   │   ├── analysis.py             [/api/analysis/* - analysis endpoints]
│   │   │   ├── backtest.py             [/api/backtest/* - backtesting]
│   │   │   └── ai_assistant.py         [AI chat endpoints]
│   │
│   ├── requirements.txt
│   ├── docker-compose.yml
│   └── .env
│
├── scripts/
│   └── fetch_realdata.py                [Standalone data fetching script]
│
├── docs/
├── frontend/
│
├── DATA_PIPELINE_ARCHITECTURE.md       [Detailed architecture]
├── DATA_PIPELINE_QUICK_REFERENCE.md    [Quick guide]
└── DATA_PIPELINE_FILE_MAP.md           [This file]

Core Pipeline Modules

1. Data Ingestion Layer

File: /root/work/stock/backend/app/modules/data_fetcher.py (355 lines)

Class: DataFetcher

Key Methods:

Method	Lines	Purpose
`__init__()`	21-25	Initialize with TuShare token
`fetch_stock_data()`	27-77	Main OHLCV fetch with TuShare/yfinance fallback
`get_realtime_quote()`	79-129	Real-time price quotes with caching
`fetch_fundamentals()`	131-196	Fundamental metrics (PE, PB, market cap)
`_fetch_tushare_data()`	197-268	TuShare API wrapper
`_convert_symbol()`	270-285	Symbol format conversion
`fetch_sector_data()`	299-319	Sector performance
`get_market_sentiment()`	321-350	Market sentiment indicators

Entry Point for Pipeline: Line 27

data_dict = data_fetcher.fetch_stock_data(symbols, period="3mo")

Data Flow:

fetch_stock_data(symbols)
    ├─ For each symbol:
    │   ├─ Check if Chinese stock (6/0/3 prefix)
    │   ├─ Try _fetch_tushare_data() → _convert_symbol("600000" → "600000.SH")
    │   ├─ Fall back to yfinance → _convert_symbol("600000" → "600000.SS")
    │   └─ Normalize columns → {Open, High, Low, Close, Volume}
    └─ Return: Dict[symbol: DataFrame]

Cache Interaction:

get_realtime_quote() checks stock:quote:{symbol} (line 92-98)
fetch_fundamentals() checks stock:fundamentals:{symbol} (line 141-146)

2. Feature Engineering Layer

File: /root/work/stock/backend/app/modules/feature_engineer.py (244 lines)

Class: FeatureEngineer

Key Methods:

Method	Lines	Purpose	# Features
`engineer_features()`	23-54	Main pipeline	33 total
`normalize_features()`	56-85	StandardScaler normalization	-
`_add_moving_averages()`	87-100	MA indicators	11
`_add_momentum_indicators()`	102-120	RSI, MACD, Stochastic	8
`_add_volatility_indicators()`	122-139	ATR, Bollinger Bands	8
`_add_trend_indicators()`	141-156	Trend & volume features	4
`_calculate_rsi()`	158-168	RSI calculation	1
`_calculate_macd()`	170-180	MACD calculation	3
`_calculate_atr()`	182-193	ATR calculation	1
`_calculate_bollinger_bands()`	195-206	Bollinger bands	4
`_calculate_stochastic()`	208-217	Stochastic oscillator	2
`get_feature_names()`	219-239	List all feature names	-

Entry Point for Pipeline: Line 23

df_features = feature_engineer.engineer_features(df)

Configuration Flags:

# app/config.py - Lines 46-49
enable_technical_features: bool = true
enable_fundamental_features: bool = true
enable_momentum_features: bool = true

Feature Categories:

Moving Averages (11):     MA5, MA10, MA20, MA50, MA200 + Ratios + Cross
Momentum (8):             RSI_14, MACD, MACD_Signal, MACD_Hist, Stoch_K/D, ROC_12, Momentum_10
Volatility (8):           ATR_14, BB_Middle/Upper/Lower/Width/Position, Volatility_20/60
Trend (4):                Momentum_5d/20d, Volume_Ratio, Avg_Return_10d
─────────────────
TOTAL:                    33 features

Output Format: (Line 73-74)

feature_cols = [col for col in df.columns if col not in ["Open", "High", "Low", "Close", "Volume"]]
X = latest_features[feature_cols].fillna(0).values  # Shape: [1, 33]

3. ML Training & Prediction Layer

File: /root/work/stock/backend/app/modules/ml_trainer.py (251 lines)

Class: MLTrainer

Key Methods:

Method	Lines	Purpose
`__init__()`	32-36	Initialize trainer
`train_model()`	38-106	Train XGBoost/LightGBM
`predict_scores()`	108-135	Get 0-100 scores
`evaluate_model()`	137-164	Calculate metrics
`get_feature_importance()`	166-198	Feature importance ranking
`save_model()`	200-209	Serialize to disk
`load_model()`	211-223	Deserialize from disk

Entry Point for Pipeline: Line 108

score = model.predict_scores(X)[0]  # Line 86 in recommendations.py

Model Architecture:

XGBoost (default - Line 60-75):
├─ objective: "binary:logistic"
├─ max_depth: 5
├─ learning_rate: 0.1
├─ n_estimators: 100
└─ Output: Binary probability [0-1]

LightGBM (alternative - Line 77-93):
├─ objective: "binary"
├─ num_leaves: 31
├─ learning_rate: 0.05
├─ n_estimators: 100
└─ Output: Binary probability [0-1]

Prediction Pipeline:

# Line 124: Get class 1 probability
proba = self.model.predict_proba(X)[:, 1]
# Line 127: Scale to 0-100
scores = proba * 100

Configuration:

# app/config.py - Lines 37-39
model_type: str = "xgboost"
model_path: str = "/app/models/stock_model.pkl"
feature_scaler_path: str = "/app/models/feature_scaler.pkl"

Dummy Model Creation: (Line 226-246)

Used when model not pre-trained
Creates synthetic 1000 samples × 33 features
Trains XGBoost for demo purposes

4. Caching Layer

File: /root/work/stock/backend/app/modules/cache_manager.py (127 lines)

Class: CacheManager

Key Methods:

Method	Lines	Purpose
`__init__()`	17-19	Initialize Redis connection
`set()`	21-41	Store with TTL
`get()`	43-65	Retrieve & parse JSON
`delete()`	67-74	Remove key
`clear_prefix()`	76-93	Bulk delete with pattern
`exists()`	95-101	Check existence
`health_check()`	103-110	Test Redis connection

Cache Keys Definition: (Lines 118-126)

CACHE_KEYS = {
    "quote": "stock:quote:{symbol}",                 # 60s
    "features": "stock:features:{symbol}",           # 300s
    "score": "stock:score:{symbol}",                 # 60s
    "fundamentals": "stock:fundamentals:{symbol}",   # 3600s
    "recommendation": "recommendation:top10:latest", # 60s
    "llm_explanation": "llm:explanation:{symbol}",   # 3600s
    "market_context": "market:context:latest",       # 60s
}

Configuration:

# app/config.py - Lines 28-29, 62-65
redis_url: str = "redis://localhost:6379/0"
cache_ttl_quote: int = 60
cache_ttl_features: int = 300
cache_ttl_recommendation: int = 60
cache_ttl_llm: int = 3600

Usage Examples in Pipeline:

From data_fetcher.py:

# Line 92-98: Check quote cache
cache_key = CACHE_KEYS["quote"].format(symbol=symbol)
cached = cache.get(cache_key)
if cached:
    quotes[symbol] = cached
    continue

From recommendations.py:

# Line 32-37: Check recommendation cache
cache_key = CACHE_KEYS["recommendation"]
cached = cache.get(cache_key)
if cached:
    return cached

5. LLM Integration

File: /root/work/stock/backend/app/modules/llm_ranker.py (296 lines)

Class: LLMRanker

Key Methods:

Method	Lines	Purpose
`__init__()`	21-27	Initialize OpenRouter client
`generate_recommendation_explanation()`	29-69	Generate top 10 explanation
`generate_stock_analysis()`	71-115	Single stock analysis
`_call_claude_api()`	117-184	API call wrapper
`_format_recommendation_prompt()`	186-221	Prompt formatting
`_format_stock_analysis_prompt()`	223-255	Prompt formatting
`_get_mock_response()`	257-262	Fallback response

Function: select_top_recommendations() (Lines 265-291)

def select_top_recommendations(scores, metadata, n=10):
    # Sort scores descending, return top N with metadata

Configuration:

# app/config.py - Lines 32, 56-59, 65
openrouter_key: str = ""
llm_model: str = "anthropic/claude-4.5-sonnet"
llm_temperature: float = 0.7
llm_max_tokens: int = 500
llm_enable_cache: bool = true
cache_ttl_llm: int = 3600

API Integration:

# Line 24: OpenRouter endpoint
self.api_url = "https://openrouter.ai/api/v1/chat/completions"

# Line 159: HTTP request
response = requests.post(self.api_url, headers=headers, json=payload)

API Routes / Endpoints

1. Recommendations Endpoint

File: /root/work/stock/backend/app/routes/recommendations.py (253 lines)

Endpoint 1: GET /api/recommendations (Lines 19-143)

Full Pipeline Call Sequence:

# Line 32-37: Check cache
cache_key = CACHE_KEYS["recommendation"]
cached = cache.get(cache_key)

# Line 40-44: Fetch data (data_fetcher module)
symbols = ["600000", "600031", "601988", "601398"]
data_dict = data_fetcher.fetch_stock_data(symbols, period="3mo")

# Line 71: Engineer features (feature_engineer module)
df_features = feature_engineer.engineer_features(df)

# Line 74-78: Get ML prediction (ml_trainer module)
feature_cols = [col for col in df_features.columns if col not in ["Open", "High", "Low", "Close", "Volume"]]
X = latest_features[feature_cols].fillna(0).values
score = model.predict_scores(X)[0]

# Line 89-90: Get metadata (data_fetcher module)
quotes = data_fetcher.get_realtime_quote([symbol])
fundamentals = data_fetcher.fetch_fundamentals(symbol)

# Line 114: Select top 10 (llm_ranker module)
recommendations = select_top_recommendations(scores, metadata, n=10)

# Line 121-125: Generate explanation (llm_ranker module)
explanation = llm_ranker.generate_recommendation_explanation(
    recommendations[:5], market_context, language="zh")

# Line 137: Cache result
cache.set(cache_key, result, ttl=60)

Endpoint 2: GET /api/recommendations/{symbol} (Lines 146-237)

Similar flow but for single stock
Returns detailed analysis + price chart

Endpoint 3: GET /api/health (Lines 240-252)

Health check endpoint
Checks cache and service status

2. Analysis Endpoints

File: /root/work/stock/backend/app/routes/analysis.py (156 lines)

Endpoints:

GET /api/analysis/sectors (Lines 14-49)
- Uses: data_fetcher.fetch_sector_data()
- Simulated sector performance scores
GET /api/analysis/signals/{symbol} (Lines 52-104)
- Returns technical signals
- Simulated signal analysis
GET /api/analysis/market (Lines 107-155)
- Uses: data_fetcher.get_market_sentiment()
- Market indices and breadth

3. Backtesting Endpoint

File: /root/work/stock/backend/app/routes/backtest.py (102 lines)

Backtester Module: /root/work/stock/backend/app/modules/backtester.py (250 lines)

Key Methods:

Method	Lines	Purpose
`backtest_strategy()`	20-104	Main backtest logic
`_execute_trades()`	133-170	Simulate trade execution
`_calculate_metrics()`	211-245	Performance metrics

Metrics Calculated: (Lines 238-245)

{
    "total_return": percentage,
    "sharpe_ratio": risk-adjusted return,
    "max_drawdown": worst peak-to-trough,
    "win_rate": percentage of profitable trades,
    "final_equity": end balance
}

Configuration & Setup

File: /root/work/stock/backend/app/config.py (91 lines)

Settings Class: Settings (Lines 12-84)

Key Configuration Variables:

Variable	Default	Line	Purpose
`app_env`	development	16	Environment
`debug`	True	17	Debug mode
`database_url`	postgres://	23-26	PostgreSQL connection
`redis_url`	redis://localhost:6379/0	29	Redis connection
`tushare_token`	-	34	TuShare API key
`openbb_token`	-	33	OpenBB API key
`stock_symbols`	600000,600031,601988,601398	42	Default watchlist
`lookback_days`	365	43	Historical data range
`enable_technical_features`	true	47	Feature engineering flag
`model_type`	xgboost	37	ML model type
`model_path`	/app/models/stock_model.pkl	38	Model file location
`llm_model`	anthropic/claude-4.5-sonnet	56	Claude model
`cache_ttl_quote`	60	62	Quote cache TTL
`cache_ttl_features`	300	63	Features cache TTL
`cors_origins`	localhost:8080	69-71	CORS allowed origins

Getter Function: (Lines 87-90)

@lru_cache()
def get_settings() -> Settings:
    return Settings()

Database Layer

File: /root/work/stock/backend/app/database.py (61 lines)

Functions:

Function	Lines	Purpose
SQLAlchemy engine	14-24	Create database engine
SessionLocal	27-31	Session factory
get_db()	37-43	Dependency injection
init_db()	46-49	Create tables
health_check()	52-60	Test connection

Configuration:

# Line 14-24
engine = create_engine(
    settings.database_url,
    pool_size=10,
    max_overflow=20,
    connect_args={"connect_timeout": 10}
)

Main Application

File: /root/work/stock/backend/main.py (160 lines)

FastAPI App Setup: (Lines 57-73)

Route Registration: (Lines 76-98)

app.include_router(
    recommendations.router,
    prefix="/api",
    tags=["Recommendations"]
)
app.include_router(analysis.router, prefix="/api", tags=["Analysis"])
app.include_router(backtest.router, prefix="/api", tags=["Backtest"])
app.include_router(ai_assistant.router, tags=["AI Assistant"])

Lifespan Management: (Lines 28-54)

Startup: Check DB & Redis health, initialize tables
Shutdown: Cleanup

Standalone Data Fetching Script

File: /root/work/stock/scripts/fetch_realdata.py (114 lines)

Purpose: Standalone script to fetch and save stock data to CSV

Main Functions:

Function	Lines	Purpose
`test_token()`	26-40	Validate TuShare token
`fetch_stock_data()`	42-64	Fetch 1-year data
`save_to_csv()`	66-74	Export to CSV files
`main()`	76-113	Orchestrate fetching

Usage:

python /root/work/stock/scripts/fetch_realdata.py
# Fetches all symbols in STOCK_SYMBOLS list
# Saves to: /root/work/stock/data/{symbol}_1y.csv

Data Flow Connections Summary

Path 1: User Request → Recommendation

User Request
    ↓
recommendations.py: get_recommendations()
    ├─ cache_manager.py: get("recommendation:top10:latest")
    ├─ data_fetcher.py: fetch_stock_data()
    │   └─ _fetch_tushare_data() or yfinance
    ├─ data_fetcher.py: get_realtime_quote()
    ├─ data_fetcher.py: fetch_fundamentals()
    ├─ feature_engineer.py: engineer_features()
    ├─ ml_trainer.py: predict_scores()
    ├─ llm_ranker.py: select_top_recommendations()
    ├─ llm_ranker.py: generate_recommendation_explanation()
    ├─ cache_manager.py: set("recommendation:top10:latest", ttl=60)
    └─ JSON Response

Path 2: Cache Hit

User Request
    ↓
recommendations.py: get_recommendations()
    ├─ cache_manager.py: get("recommendation:top10:latest")
    └─ Return Cached (< 10ms)

Path 3: Single Stock Detail

User Request: /api/recommendations/{symbol}
    ↓
recommendations.py: get_stock_detail({symbol})
    ├─ data_fetcher.py: fetch_stock_data([symbol], period="1y")
    ├─ feature_engineer.py: engineer_features()
    ├─ data_fetcher.py: fetch_fundamentals()
    ├─ data_fetcher.py: get_realtime_quote()
    ├─ llm_ranker.py: generate_stock_analysis()
    └─ JSON Response with chart

Key Data Types and Transformations

Stage 1: Raw OHLCV

# From data_fetcher.py
type: pd.DataFrame
shape: (252, 5)  # rows, columns
columns: [Open, High, Low, Close, Volume]
index: DatetimeIndex
dtypes: [float, float, float, float, int]
memory: ~40-50 KB

Stage 2: Features

# From feature_engineer.py
type: pd.DataFrame
shape: (252, 38)  # 5 OHLCV + 33 features
columns: [Open, High, Low, Close, Volume, MA5, ..., Avg_Return_10d]
index: DatetimeIndex
dtypes: [float, ...]
memory: ~100-150 KB

Stage 3: Model Input

# From ml_trainer.py
type: np.ndarray
shape: (1, 33)  # 1 sample, 33 features only
dtype: float32
normalized: Z-score [-3, +3]
memory: < 1 KB

Stage 4: Prediction

# From ml_trainer.py
type: np.ndarray
shape: (4,)  # 4 stocks
dtype: float
range: [0, 100]
memory: < 1 KB

Stage 5: Recommendation

# From recommendations.py
type: List[Dict]
keys: [symbol, score, name, current_price, industry, pe_ratio, 
       pb_ratio, predicted_return, confidence, technical_signals]
length: 10 (top recommendations)
size: ~5-10 KB JSON

Important Constants and Settings

Default Symbols

STOCK_SYMBOLS = ["600000", "600031", "601988", "601398"]
# 600000: SPDB (Shanghai Pudong Development Bank)
# 600031: Shenergy Company
# 601988: China Bank
# 601398: ICBC (Industrial and Commercial Bank of China)

Feature Engineering Flags

ENABLE_TECHNICAL_FEATURES = True      # Generate all 33 features
ENABLE_FUNDAMENTAL_FEATURES = True    # Include fundamental metrics
ENABLE_MOMENTUM_FEATURES = True       # Include momentum indicators

ML Model Parameters

XGBoost:
  max_depth = 5
  learning_rate = 0.1
  n_estimators = 100
  objective = "binary:logistic"

Cache TTLs

CACHE_TTL_QUOTE = 60                  # 1 minute
CACHE_TTL_FEATURES = 300              # 5 minutes
CACHE_TTL_RECOMMENDATION = 60         # 1 minute
CACHE_TTL_LLM = 3600                  # 1 hour

Troubleshooting: Tracing Data Through Modules

Problem: No recommendations returned

Trace path:

Check recommendations.py:get_recommendations() (line 20)
Check data_fetcher.py:fetch_stock_data() (line 27)
- Are symbols correct?
- Is TuShare token valid?
- Can yfinance reach data?
Check feature_engineer.py:engineer_features() (line 23)
- Is DataFrame structure correct?
Check ml_trainer.py:predict_scores() (line 108)
- Is model loaded?
Check llm_ranker.py:select_top_recommendations() (line 265)

Problem: Slow recommendations

Check cache:

cache_manager.py:health_check() (line 103)
Monitor CACHE_KEYS["recommendation"] hits

Check pipeline:

Data fetch latency: data_fetcher.py
Feature engineering time: feature_engineer.py
Model inference time: ml_trainer.py
LLM API latency: llm_ranker.py

Problem: Invalid feature values

Check sequence:

data_fetcher.py - validate OHLCV columns
feature_engineer.py - check NaN handling
ml_trainer.py - verify normalization

20 KiB Raw Permalink Blame History Unescape Escape

Data Pipeline - File Map and Code Locations

Directory Structure

Core Pipeline Modules

1. Data Ingestion Layer

2. Feature Engineering Layer

3. ML Training & Prediction Layer

4. Caching Layer

5. LLM Integration

API Routes / Endpoints

1. Recommendations Endpoint

2. Analysis Endpoints

3. Backtesting Endpoint

Configuration & Setup

Database Layer

Main Application

Standalone Data Fetching Script

Data Flow Connections Summary

Path 1: User Request → Recommendation

Path 2: Cache Hit

Path 3: Single Stock Detail

Key Data Types and Transformations

Stage 1: Raw OHLCV

Stage 2: Features

Stage 3: Model Input

Stage 4: Prediction

Stage 5: Recommendation

Important Constants and Settings

Default Symbols

Feature Engineering Flags

ML Model Parameters

Cache TTLs

Troubleshooting: Tracing Data Through Modules

Problem: No recommendations returned

Problem: Slow recommendations

Problem: Invalid feature values

20 KiB

Raw Permalink Blame History