ljwx-stock/CODEBASE_ANALYSIS.md

Stock Recommendation System - Complete Codebase Analysis

Executive Summary

The stock recommendation system is a data-driven AI platform that recommends stocks using machine learning and Claude API for explanations. Phase 1 is complete (real TuShare data integration), and the system is ready for Phase 2 and Phase 3 implementation.

Current Status: Production-ready MVP with fundamental data pipeline Completion Date: November 10, 2025

---

Part 1: Phase 2 Implementation Plan (Market Sentiment + Announcements)

Phase 2 Objectives

Enable the system to incorporate market sentiment and real-time announcements into recommendation generation. This adds critical context for stock predictions beyond technical and fundamental analysis.

Phase 2 Scope

Component	Purpose	Status
Sentiment Fetcher	Fetch market sentiment from Sina, Tencent, East Money	Not Started
Announcement Crawler	Extract corporate announcements and news	Not Started
Sentiment Analyzer	NLP analysis of market mood (bullish/bearish)	Not Started
Announcement Parser	Extract key information from announcements	Not Started
Integration Layer	Combine sentiment + announcements with existing data	Not Started

Phase 2 Data Sources

1. Market Sentiment Sources:

   • Sina Finance API (市场情绪指数)
   • Tencent Stock (腾讯股票)
   • East Money (东财数据)
   • Baidu Index (百度指数)
   • Stock forum sentiment analysis

2. Announcement Sources:

   • TuShare Announcement API (公告信息)
   • Stock Exchange announcements (交易所公告)
   • Company press releases (公司新闻)
   • Industry news aggregation

3. News Sources:

   • Sina Finance News
   • CNBC/Reuters (international)
   • Industry-specific publications

Phase 2 New Modules to Create

1. sentiment_fetcher.py (~300 lines)

Purpose: Fetch and aggregate market sentiment from multiple sources

class SentimentFetcher:
    def get_overall_sentiment() -> Dict:
        # Returns: {overall: "bullish|neutral|bearish", strength: 0-1, volatility: 0-1}
        pass
    
    def get_sector_sentiment(sector: str) -> Dict:
        # Returns sentiment for specific sectors (金融, 医药, 消费, etc.)
        pass
    
    def get_stock_sentiment(symbol: str) -> Dict:
        # Returns sentiment for specific stock from social media
        pass

Data Flow:

• Poll Sina Finance API for market index sentiment
• Analyze trading volume patterns
• Parse social media mentions (Weibo, stock forums)
• Calculate sentiment score (0-100)
• Cache results (TTL: 5 minutes)

2. announcement_crawler.py (~350 lines)

Purpose: Fetch and parse corporate announcements

class AnnouncementCrawler:
    def fetch_stock_announcements(symbol: str, days: int = 30) -> List[Dict]:
        # Returns: [{"date": "2025-11-10", "title": "...", "impact": "high|medium|low"}]
        pass
    
    def parse_announcement(content: str) -> Dict:
        # Extract key information: funding, contracts, earnings, etc.
        pass
    
    def calculate_impact_score(announcement: Dict) -> float:
        # Score announcement impact on stock (0-100)
        pass

Data Sources:

• TuShare API: pro.news(), pro.disclosed_fund()
• Stock Exchange APIs (SSE, SZSE)
• Company official websites

3. sentiment_analyzer.py (~250 lines)

Purpose: NLP-based sentiment analysis and extraction

class SentimentAnalyzer:
    def analyze_text_sentiment(text: str) -> Dict:
        # Returns: {sentiment: "positive|negative|neutral", confidence: 0-1, keywords: [...]}
        pass
    
    def extract_keywords(text: str) -> List[str]:
        # Extract important keywords from announcement
        pass
    
    def score_announcement(text: str, symbol: str) -> float:
        # Score relevance and impact for specific stock
        pass

Technologies:

• Chinese NLP library: jieba for word segmentation
• Sentiment classification: DistilBERT or local LLaMA model
• Keyword extraction: TF-IDF or Topic Modeling

4. market_context_builder.py (~200 lines)

Purpose: Aggregate all market context data

class MarketContextBuilder:
    def build_market_context() -> Dict:
        # Combines sentiment + announcements + technical analysis
        return {
            "overall_sentiment": "...",
            "sector_sentiment": {...},
            "top_positive_news": [...],
            "top_negative_news": [...],
            "implied_sectors": [...],
            "risk_level": "low|medium|high"
        }

Phase 2 API Changes

New Endpoints

# GET /api/market/sentiment
{
    "overall": "bullish",
    "strength": 0.72,
    "sectors": {
        "金融": 0.65,
        "医药": 0.78,
        "房地产": 0.45
    },
    "timestamp": "2025-11-10T14:30:00Z"
}

# GET /api/announcements/{symbol}
{
    "announcements": [
        {
            "date": "2025-11-10",
            "title": "发布三季度财报",
            "impact": "high",
            "summary": "...",
            "impact_score": 85
        }
    ]
}

# Updated GET /api/recommendations
{
    "recommendations": [...],
    "market_context": {
        "sentiment": "bullish",
        "key_drivers": ["金融股走强", "科技股反弹"],
        "risk_alerts": ["房地产风险上升"],
        "announcement_impact": {...}
    }
}

Updated AI Chat

# AI Assistant will now consider:
# 1. Technical signals (existing)
# 2. Fundamental metrics (Phase 1)
# 3. Market sentiment (Phase 2 new)
# 4. Recent announcements (Phase 2 new)

@router.post("/api/ai/chat")
async def chat(request: ChatRequest):
    # Now includes market sentiment context when answering
    # e.g., "600000当前在金融板块走强的背景下..."

Phase 2 Database Additions

class MarketSentiment(Base):
    timestamp: datetime
    overall_sentiment: str  # bullish, neutral, bearish
    sentiment_score: float  # 0-100
    strength: float  # 0-1
    volatility: float  # 0-1
    volume: str  # high, medium, low
    sector_sentiments: JSON  # {sector_name: score}

class StockAnnouncement(Base):
    symbol: str
    date: date
    title: str
    content: str
    announcement_type: str  # earnings, financing, contract, etc.
    impact_score: float  # 0-100
    sentiment: str  # positive, negative, neutral
    extracted_info: JSON  # key data from announcement

class NewsItem(Base):
    symbol: str
    date: datetime
    title: str
    source: str
    sentiment: str
    relevance_score: float

Phase 2 Implementation Schedule

Week	Task	Hours	Deliverable
1	Build sentiment_fetcher + tests	8	Sentiment data pipeline
2	Build announcement_crawler + parser	8	Announcement extraction
3	Integrate sentiment into recommendations	8	Updated recommendation logic
4	Integration testing + documentation	4	Phase 2 complete

Estimated Total: ~28 hours

---

Part 2: Phase 3 Implementation Plan (Performance Optimization & Knowledge Base)

Phase 3 Objectives

Optimize system performance, build an enterprise knowledge base, and prepare for production deployment at scale.

Phase 3 Components

Component	Purpose	Status
Query Optimization	Faster data fetching and caching	Not Started
Model Caching	Pre-load models for instant predictions	Not Started
Knowledge Base	Store investor research and analysis	Not Started
Advanced Caching	Redis optimization, multi-level cache	Not Started
Monitoring	Performance metrics and alerting	Not Started
Load Testing	Stress testing and capacity planning	Not Started

Phase 3 New Modules

1. knowledge_base.py (~400 lines)

Purpose: Vector database for storing and retrieving investment insights

class KnowledgeBase:
    def __init__(self):
        # Embeddings: using OpenAI embeddings or local model
        # Vector DB: Pinecone, Weaviate, or Milvus
        pass
    
    def add_research_document(doc: str, metadata: Dict) -> str:
        # Embed and store document (analyst reports, investment theses)
        pass
    
    def query_similar_insights(query: str, top_k: int = 5) -> List[Dict]:
        # Find similar historical analysis for context
        pass
    
    def cache_company_profiles(symbols: List[str]):
        # Pre-cache comprehensive company information
        pass

Content Sources:

• Analyst reports (from financial platforms)
• Historical recommendation analysis
• Industry research documents
• Company earnings call transcripts

2. cache_optimizer.py (~300 lines)

Purpose: Advanced multi-level caching strategy

class CacheOptimizer:
    def __init__(self):
        self.l1_cache = InMemoryCache()  # FastAPI memory
        self.l2_cache = RedisCache()     # Redis
        self.l3_cache = SQLiteCache()    # Persistent local
        pass
    
    def get_with_fallback(key: str) -> Optional[Any]:
        # Check L1 -> L2 -> L3 -> Fetch fresh data
        pass
    
    def prefetch_common_queries():
        # Pre-warm cache with popular stocks
        pass
    
    def adaptive_ttl(key: str) -> int:
        # Dynamically adjust TTL based on volatility
        pass

Cache Strategy:

• L1: In-memory cache (TTL: 1-5 minutes, high hit rate for real-time queries)
• L2: Redis (TTL: 15 minutes - 1 hour, shared across instances)
• L3: SQLite (TTL: 1-7 days, persistent backup)
• L4: Original data sources (fetch only on miss)

3. performance_monitor.py (~250 lines)

Purpose: Track system performance and identify bottlenecks

class PerformanceMonitor:
    def track_api_latency(endpoint: str, duration_ms: float):
        # Track endpoint response times
        pass
    
    def track_cache_hit_rate(cache_layer: str, hit: bool):
        # Monitor cache effectiveness
        pass
    
    def track_data_fetch_time(source: str, duration_ms: float):
        # Monitor data source performance
        pass
    
    def generate_performance_report() -> Dict:
        # Daily/weekly performance summary
        pass

Metrics Tracked:

• API endpoint latencies (target: <200ms for cached, <3s for fresh)
• Cache hit rates (target: >80% for quotes, >60% for features)
• Data source latencies (target: <2s for TuShare, <1s for sentiment)
• Model prediction time (target: <500ms)
• Memory usage (target: <500MB baseline)

4. batch_processor.py (~280 lines)

Purpose: Efficient batch processing of data updates

class BatchProcessor:
    def batch_fetch_fundamentals(symbols: List[str]) -> Dict:
        # Fetch multiple stocks in parallel with rate limiting
        pass
    
    def batch_generate_recommendations(n_stocks: int = 100) -> List[Dict]:
        # Generate recommendations for multiple stocks at once
        pass
    
    def batch_update_cache(updates: List[Tuple]):
        # Atomic batch cache updates
        pass

Phase 3 Architecture Improvements

Current Architecture (Phase 1-2)

User Request
    ↓
FastAPI Endpoint
    ↓
Redis Cache (check)
    ↓ (miss)
Data Fetcher
    ↓
ML Model
    ↓
LLM Ranker
    ↓
Response

Optimized Architecture (Phase 3)

User Request
    ↓
API Gateway + Load Balancer
    ↓
L1 Cache (In-Memory)
    ↓ (miss)
L2 Cache (Redis Cluster)
    ↓ (miss)
Worker Pool (Parallel Fetching)
    ├─ Data Fetcher (async)
    ├─ Model Predictor (batched)
    └─ Knowledge Base Query (vectorized)
    ↓
Result Aggregator
    ↓
L2 Cache (write)
    ↓
Response Formatter
    ↓
Response

Phase 3 Database Optimizations

class OptimizedSchema:
    # Add indexes for frequently accessed fields
    class StockQuote(Base):
        __table_args__ = (
            Index('idx_symbol_timestamp', 'symbol', 'timestamp'),
            Index('idx_timestamp', 'timestamp'),
            Index('idx_score', 'score'),
        )
    
    # Materialized views for aggregations
    class DailyRecommendationCache(Base):
        # Pre-calculated recommendations, updated daily
        date: date
        top_10: JSON  # Pre-serialized top 10
        market_context: JSON
    
    # Partitioning by date for large tables
    class HistoricalPrices(Base):
        __table_args__ = {
            'mysql_partition_by': 'RANGE (YEAR(trade_date))'
        }

Phase 3 Infrastructure

Deployment Architecture

┌─────────────────────────────────────────┐
│         Nginx / Load Balancer           │
└────────────────┬────────────────────────┘
                 │
    ┌────────────┼────────────┐
    │            │            │
┌───▼──┐    ┌───▼──┐    ┌───▼──┐
│App 1 │    │App 2 │    │App 3 │  (FastAPI replicas)
└───┬──┘    └───┬──┘    └───┬──┘
    │           │           │
    └───────────┼───────────┘
                │
    ┌───────────┼──────────────┐
    │           │              │
┌───▼──┐   ┌───▼────┐    ┌────▼──┐
│Redis │   │Postgres│    │SQLite │ (local cache)
│Cluster  │           │
└────────┘   └────────┘    └───────┘

Kubernetes Deployment (Optional)

apiVersion: apps/v1
kind: Deployment
metadata:
  name: stock-recommendation-api
spec:
  replicas: 3
  resources:
    requests:
      cpu: 500m
      memory: 512Mi
    limits:
      cpu: 1000m
      memory: 1Gi
  volumeMounts:
    - name: model-cache
      mountPath: /app/models

Phase 3 Monitoring & Alerting

Key Metrics

# Prometheus metrics
api_latency_histogram = Histogram('api_latency_ms', ...)
cache_hit_rate = Gauge('cache_hit_rate', ...)
model_prediction_time = Histogram('model_prediction_ms', ...)
data_fetch_errors = Counter('data_fetch_errors', ...)

# Alerts (via AlertManager)
- API p99 latency > 5s
- Cache hit rate < 50%
- Error rate > 1%
- Memory usage > 80%

Phase 3 Testing Strategy

Load Testing

# Simulate 100 concurrent users, 1000 requests/second
k6 run load_test.js --vus 100 --duration 5m

Performance Benchmarks

Target metrics:
- Recommendation generation: <500ms (p95)
- Stock analysis: <1s (p95)
- Market sentiment fetch: <2s (p95)
- Cache hit rate: >80% for quotes
- System throughput: 1000 req/s

Phase 3 Implementation Schedule

Week	Task	Hours	Deliverable
1	Implement cache optimization	8	Multi-level cache
2	Build knowledge base	8	Vector DB integration
3	Add monitoring & alerting	6	Prometheus + Grafana
4	Load testing & optimization	8	Performance report
5	Documentation & deployment	4	Phase 3 complete

Estimated Total: ~34 hours

---

Part 3: Current Module Structure & Architecture

Existing Modules Overview

1. data_fetcher.py (355 lines)

Location: /root/work/stock/backend/app/modules/data_fetcher.py

Responsibilities:

• Fetch historical OHLCV (Open, High, Low, Close, Volume) data
• Get real-time stock quotes
• Fetch fundamental data via TuShare API
• Support multiple data sources (TuShare, yfinance)

Key Classes:

class DataFetcher:
    def fetch_stock_data(symbols: List[str], period: str, interval: str) -> Dict[str, pd.DataFrame]
    def get_realtime_quote(symbols: List[str]) -> Dict[str, Dict]
    def fetch_fundamentals(symbol: str) -> Dict
    def fetch_sector_data() -> Dict[str, float]
    def get_market_sentiment() -> Dict

Data Sources:

• TuShare API (Chinese A-stocks priority)
• yfinance fallback (for international stocks)
• Internal simulation (for market sentiment)

Features:

• Automatic source selection (TuShare for CN, yfinance for others)
• Error handling with graceful degradation
• Returns data in standardized DataFrame format

2. feature_engineer.py (244 lines)

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

Responsibilities:

• Engineer technical indicators from raw OHLCV data
• Normalize features for ML models
• Generate 40+ technical features

Key Classes:

class FeatureEngineer:
    def engineer_features(df: pd.DataFrame) -> pd.DataFrame
    def normalize_features(df: pd.DataFrame, scaler, fit: bool) -> Tuple[pd.DataFrame, Scaler]
    def get_feature_names() -> List[str]

Features Generated (40 total):

• Moving Averages: MA5, MA10, MA20, MA50, MA200 + ratios + crosses
• Momentum: RSI(14), MACD, Stochastic, ROC(12), Momentum(10)
• Volatility: ATR(14), Bollinger Bands (4 features), Historical volatility
• Trend: Momentum(5d/20d), Volume ratios, Average returns
• Advanced: Price position in BB, Cross signals

Workflow:

Raw OHLCV → Moving Averages → Momentum Indicators → 
Volatility Indicators → Trend Indicators → Normalized Features → ML Model

3. ml_trainer.py (251 lines)

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

Responsibilities:

• Train ML models (XGBoost, LightGBM)
• Make stock score predictions (0-100)
• Evaluate model performance
• Manage model persistence

Key Classes:

class MLTrainer:
    def train_model(X_train, y_train, model_type: str) -> Model
    def predict_scores(X: np.ndarray, normalize: bool = True) -> np.ndarray
    def evaluate_model(y_true, y_pred, y_proba) -> Dict[str, float]
    def get_feature_importance(top_n: int = 20) -> Dict[str, float]
    def save_model(path: str) -> bool
    def load_model(path: str) -> bool

Models Supported:

• XGBoost: Better accuracy, slower training
• LightGBM: Faster, lower memory, good accuracy

Label Definition:

• y=1: Stock will return >3% in next 5 days
• y=0: Otherwise

Metrics:

• Accuracy, Precision, Recall, F1-Score, ROC-AUC

4. llm_ranker.py (296 lines)

Location: /root/work/stock/backend/app/modules/llm_ranker.py

Responsibilities:

• Generate natural language explanations for recommendations
• Analyze individual stocks using Claude API
• Select top N recommendations
• Cache LLM responses

Key Classes:

class LLMRanker:
    def generate_recommendation_explanation(top_stocks, market_context, language) -> str
    def generate_stock_analysis(symbol, stock_data, technical_signals) -> str
    
def select_top_recommendations(scores: Dict, metadata: Dict, n: int) -> List[Dict]

Features:

• Claude API integration via OpenRouter
• Smart prompt engineering for stock analysis
• Fallback mock responses when API unavailable
• Caching with configurable TTL
• Chinese/English language support

Prompt Strategy:

• System: Establishes analyst persona (20 years experience)
• User: Provides stock data + technical signals
• Output: Professional, actionable analysis (200-300 words)

5. fundamental_data.py (468 lines)

Location: /root/work/stock/backend/app/modules/fundamental_data.py

Responsibilities [Phase 1 Addition]:

• Fetch real fundamental financial data from TuShare
• Calculate derived metrics (ROE, debt ratio, PEG)
• Provide AI-friendly formatted context
• Cache all data with appropriate TTLs

Key Classes:

class FundamentalDataFetcher:
    def get_stock_fundamentals(symbol: str, date: Optional[str]) -> Dict
    def get_fundamentals_for_ai_context(symbol: str) -> Dict
    def get_multiple_stocks_fundamentals(symbols: List[str]) -> Dict[str, Dict]
    
    # Private methods for data fetching
    def _fetch_daily_basic(symbol: str, date: Optional[str]) -> Dict
    def _fetch_income_statement(symbol: str) -> Dict
    def _fetch_balance_sheet(symbol: str) -> Dict
    def _calculate_derived_metrics(data: Dict) -> Dict

Data Fetched:

• PE ratio, PB ratio, Market cap (from daily_basic)
• Revenue, Net profit, Growth rate (from income statement)
• Assets, Liabilities, Equity (from balance sheet)
• Derived: ROE, Debt ratio, PEG ratio

Interpretation Layer:

• Automatically scores metrics (低估/合理/高估)
• Human-readable evaluation (优秀/良好/需改善)
• Comparative analysis ready

6. backtester.py (250 lines)

Location: /root/work/stock/backend/app/modules/backtester.py

Responsibilities:

• Validate recommendation strategy with historical data
• Calculate performance metrics
• Simulate trading based on recommendations

Key Classes:

class Backtester:
    def backtest_strategy(recommendations_history, price_history, rebalance_interval, top_n) -> Dict
    
    # Private helpers
    def _get_recommendations_at_date(recommendations_history, date) -> List[Dict]
    def _execute_trades(recommendations, price_history, date, available_capital) -> List[Dict]
    def _calculate_pnl(trades, price_history, current_date) -> float
    def _calculate_metrics(equity_curve, daily_returns) -> Dict

Metrics Calculated:

• Total Return (%)
• Sharpe Ratio (risk-adjusted return)
• Maximum Drawdown (worst peak-to-trough decline)
• Win Rate (% of positive days)
• Final Equity

Trading Simulation:

• Equal-weight portfolio allocation
• 0.1% commission per trade
• Rebalancing at intervals (daily/weekly/monthly)

7. cache_manager.py (127 lines)

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

Responsibilities:

• Manage Redis caching for all data types
• Handle TTL (Time To Live) management
• Provide consistent cache key naming

Key Classes:

class CacheManager:
    def set(key: str, value: Any, ttl: int = 300) -> bool
    def get(key: str) -> Optional[Any]
    def delete(key: str) -> bool
    def clear_prefix(prefix: str) -> int
    def exists(key: str) -> bool
    def health_check() -> bool

Cache Key Patterns (with TTLs):

• stock:quote:{symbol} - 60s
• stock:features:{symbol} - 300s (5 min)
• stock:score:{symbol} - 60s
• stock:fundamentals:{symbol} - 3600s (1 hour)
• recommendation:top10:latest - 60s
• llm:explanation:{symbol} - 3600s
• market:context:latest - 60s

Features:

• Automatic JSON serialization
• Connection pooling
• Error handling with logging

8. openbb_provider.py (384 lines)

Location: /root/work/stock/backend/app/modules/openbb_provider.py

Responsibilities:

• Alternative data provider integration (OpenBB SDK)
• Fetch stock data from OpenBB platform
• Support for multiple stock markets

Key Classes:

class OpenBBProvider:
    def fetch_stock_data(symbols: List[str], period: str) -> Dict[str, pd.DataFrame]
    def get_technical_analysis(symbol: str, period: str) -> Dict
    def get_company_info(symbol: str) -> Dict
    def fetch_multiple_sources() -> Dict

Features:

• Fallback data source when TuShare unavailable
• Support for international stocks
• Caching integration
• Error recovery

---

Current Data Flow & Architecture

Complete Data Pipeline

┌─────────────────────────────────────────────────────────────────┐
│                      USER INTERACTION LAYER                      │
│  ┌────────────────────────────────────────────────────────────┐  │
│  │  Vue 3 Frontend                                            │  │
│  │  - Dashboard with recommendations                          │  │
│  │  - Real-time WebSocket updates                           │  │
│  │  - AI Assistant chat interface                           │  │
│  └────────────────────────────────────────────────────────────┘  │
└────────────┬───────────────────────────────────────────────────┘
             │ HTTP / WebSocket
┌────────────▼───────────────────────────────────────────────────┐
│                    FASTAPI BACKEND SERVICE                       │
│  ┌────────────────────────────────────────────────────────────┐  │
│  │  API Routes:                                               │  │
│  │  - /api/recommendations (GET) - Top 10 stocks              │  │
│  │  - /api/recommendations/{symbol} (GET) - Stock detail     │  │
│  │  - /api/analysis/{symbol} (POST) - Technical analysis    │  │
│  │  - /api/backtest (POST) - Historical validation           │  │
│  │  - /api/ai/chat (POST) - AI assistant                    │  │
│  │  - /api/ai/health (GET) - Health check                   │  │
│  └────────────────────────────────────────────────────────────┘  │
└────────────┬───────────────────────────────────────────────────┘
             │
    ┌────────┼────────┐
    │        │        │
    ▼        ▼        ▼
┌───────┐ ┌─────┐ ┌───────────┐
│Cache  │ │Data │ │LLM Ranker │
│(Redis)│ │Fetch│ │(Claude)   │
└───┬───┘ └─┬───┘ └─────┬─────┘
    │       │           │
    │ ┌─────▼───────────────────────────────────────────────┐
    │ │         DATA PROCESSING PIPELINE                    │
    │ └─────────────┬─────────────────────────────────────┘
    │               │
    │   ┌───────────┼───────────┐
    │   │           │           │
    │   ▼           ▼           ▼
    │ ┌──────┐  ┌──────────┐  ┌──────────┐
    │ │Fetch │  │Engineer  │  │ML Model  │
    │ │Data  │→ │Features  │→ │Trainer   │
    │ └──────┘  └──────────┘  └──────┬───┘
    │                                │
    │   ┌──────────────────────────────┘
    │   │
    │   ▼
    │ ┌──────────────────────────────┐
    │ │Score Prediction (0-100)      │
    │ └──────────────────────────────┘
    │                │
    └────────────────┼──────────────────────────────────┐
                     │                                  │
                     ▼                                  ▼
              ┌─────────────┐                   ┌──────────────┐
              │Select Top 10│                   │LLM Generate  │
              │Recommendations                  │Explanations  │
              └──────┬──────┘                   └──────┬───────┘
                     │                                │
                     └────────────┬────────────────────┘
                                  │
                                  ▼
                        ┌──────────────────────┐
                        │Aggregate Results     │
                        │Market Context        │
                        │Timestamp             │
                        └──────────┬───────────┘
                                  │
                                  ▼
                        ┌──────────────────────┐
                        │Response to User      │
                        │(JSON Format)         │
                        └──────────────────────┘

Specific Data Flows

1. Recommendation Generation Flow

Request: /api/recommendations
    ↓
[Check Cache] → Return cached if <60s old
    ↓ (miss)
[Data Fetcher]
    ├─ Fetch OHLCV (1 year) for 100 stocks
    └─ Cache for 5 minutes
    ↓
[Feature Engineer]
    ├─ Calculate 40 technical features
    └─ Normalize features
    ↓
[ML Model]
    ├─ Load trained model from disk
    ├─ Predict scores (0-100) for each stock
    └─ Cache scores
    ↓
[LLM Ranker]
    ├─ Select top 10 by score
    ├─ Get market context
    ├─ Call Claude API for explanation
    └─ Cache explanation for 1 hour
    ↓
[Response Formatter]
    ├─ Combine scores + explanation + context
    └─ Return JSON to frontend

2. AI Assistant Chat Flow

Request: /api/ai/chat (symbol: "600000", query: "PE怎么样?")
    ↓
[Get Fundamental Fetcher]
    └─ Initialize TuShare connection
    ↓
[Check Cache] (key: fundamental_data:600000:latest)
    ↓ (miss)
[Fetch Fundamentals]
    ├─ daily_basic: PE, PB, Market cap
    ├─ income: Revenue, Net profit, Growth rate
    └─ balance: Assets, Liabilities, Equity
    ↓
[Calculate Derived Metrics]
    ├─ ROE = Net Income / Equity
    ├─ Debt Ratio = Liabilities / Assets
    └─ PEG = PE / Growth Rate
    ↓
[Generate Interpretation]
    ├─ PE < 15 → "低估"
    ├─ ROE > 15% → "优秀"
    └─ Growth > 20% → "高速增长"
    ↓
[Format Response]
    └─ Return analysis with real values

3. Stock Detail Analysis Flow

Request: /api/recommendations/600000
    ↓
[Fetch Real-time Quote]
    └─ Current price, bid/ask, volume
    ↓
[Get Technical Features]
    └─ Latest MA, RSI, MACD, ATR, etc.
    ↓
[Extract Technical Signals]
    ├─ MA5 > MA20? → "Golden Cross"
    ├─ RSI > 70? → "Overbought"
    └─ MACD Positive Divergence?
    ↓
[Get Fundamental Data]
    └─ PE, ROE, Growth rate, etc. (cached)
    ↓
[Generate LLM Analysis]
    └─ Combine all signals for narrative
    ↓
[Return Detailed Response]
    └─ Technical + Fundamental + Narrative

---

Current Module Interactions

Module Dependency Graph

ai_assistant.py
    ├─ fundamental_data.py ──→ TuShare API
    ├─ cache_manager.py
    └─ config.py

recommendations.py
    ├─ data_fetcher.py ────→ TuShare / yfinance
    ├─ feature_engineer.py
    ├─ ml_trainer.py
    ├─ llm_ranker.py ──────→ Claude API
    ├─ cache_manager.py
    └─ config.py

backtest.py
    ├─ backtester.py
    ├─ data_fetcher.py
    ├─ feature_engineer.py
    ├─ ml_trainer.py
    └─ cache_manager.py

analysis.py
    ├─ data_fetcher.py
    ├─ feature_engineer.py
    ├─ llm_ranker.py
    └─ cache_manager.py

Configuration Management

All modules read settings from config.py:

# Environment-based configuration
DATABASE_URL → PostgreSQL connection
REDIS_URL → Redis connection
OPENROUTER_KEY → Claude API access
TUSHARE_TOKEN → Stock data API
OPENBB_TOKEN → Alternative data source

# Feature flags
ENABLE_TECHNICAL_FEATURES (default: true)
ENABLE_FUNDAMENTAL_FEATURES (default: true)
ENABLE_MOMENTUM_FEATURES (default: true)

# Model settings
MODEL_TYPE: "xgboost" or "lightgbm"
TOP_N_RECOMMENDATIONS: 10
MIN_CONFIDENCE: 0.5

# Caching
CACHE_TTL_QUOTE: 60s
CACHE_TTL_FEATURES: 300s
CACHE_TTL_RECOMMENDATION: 60s
CACHE_TTL_LLM: 3600s

---

Part 4: AI Assistant Implementation Details

AI Assistant Architecture

File: /root/work/stock/backend/app/routes/ai_assistant.py (397 lines)

Core Features

1. Real Data Integration

The AI assistant now fetches real financial data when a stock symbol is provided:

@router.post("/api/ai/chat")
async def chat(request: ChatRequest) -> ChatResponse:
    # If symbol provided, fetch real data
    if request.symbol:
        real_data_context = fetcher.get_fundamentals_for_ai_context(request.symbol)
        # Use real data in response generation

Data Retrieved:

• PE Ratio (TTM)
• PB Ratio (Price-to-Book)
• ROE (Return on Equity)
• Net Profit Growth Rate
• Debt Ratio
• PEG Ratio
• Market Cap (Total & Circulation)

2. Context-Aware Responses

The assistant generates different responses based on user query:

if "pe" in query_lower or "市盈率" in query_lower:
    if real_data and metrics.get('pe_ttm'):
        # Return PE-specific analysis with real value
        return formatted_pe_analysis(pe_ttm=7.97, status="低估")
elif "roe" in query_lower or "净资产收益率" in query_lower:
    if real_data and metrics.get('roe'):
        # Return ROE-specific analysis
        return formatted_roe_analysis(roe=4.66, status="需改善")
# ... more patterns ...

Response Templates (300-500 words each):

1. PE Analysis Template:

   • Current PE value
   • PE status assessment
   • PE ratio explanation
   • Reference standards
   • Investment advice

2. ROE Analysis Template:

   • Current ROE value
   • ROE status assessment
   • Meaning explanation
   • Performance standards
   • Investment recommendations

3. Growth Rate Template:

   • Current growth rate
   • Growth status
   • Trend interpretation
   • Reference standards
   • Risk warnings

3. Fallback Strategy

When no symbol provided or data fetch fails:

# General educational responses for common questions
if "pe" in query_lower:
    return general_pe_explanation()  # Generic PE education
elif "risk" in query_lower:
    return risk_management_guide()
elif "macd" in query_lower:
    return macd_indicator_guide()
# ... more generic topics ...

Response Structure

class ChatResponse(BaseModel):
    response: str  # Main analysis text
    thinking: str = ""  # Optional internal reasoning

Response Format (Markdown):

根据实时数据分析（数据来自 TuShare API）：

**600000 的 PE 比率分析：**

📊 **当前数据：**
- PE (TTM): 7.97
- 评估状态: 低估 (PE < 15)

**PE比率的含义：**
[Explanation of concept]

**参考标准：**
[Reference benchmarks]

**建议：**
[Investment advice]

Data Interpretation Engine

The assistant interprets financial metrics:

def _generate_interpretation(data: Dict) -> Dict:
    interpretation = {}
    
    # PE Ratio
    pe = data.get('pe_ttm')
    if pe < 15:
        interpretation['pe_status'] = '低估 (PE < 15)'
    elif pe < 25:
        interpretation['pe_status'] = '合理 (15 ≤ PE < 25)'
    else:
        interpretation['pe_status'] = '高估 (PE ≥ 25)'
    
    # ROE
    roe = data.get('roe')
    if roe > 15:
        interpretation['roe_status'] = '优秀 (ROE > 15%)'
    # ... more metrics ...
    
    return interpretation

Caching Strategy

cache_key = f"fundamental_data:{symbol}:latest"
cached_data = cache_manager.get(cache_key)

if cached_data:
    return cached_data  # Hit: <100ms

# Miss: Fetch from API
fresh_data = fetcher.get_fundamentals(symbol)
cache_manager.set(cache_key, fresh_data, ttl=86400)  # 24-hour cache
return fresh_data

Performance Impact:

• Cache hit: ~100ms response time
• Cache miss: ~1.5-2s (includes API calls)
• Hit rate target: >80% for common stocks

Error Handling

try:
    real_data_context = fetcher.get_fundamentals_for_ai_context(symbol)
except Exception as e:
    logger.warning(f"⚠️ Could not fetch real data: {e}")
    # Fall back to general response
    real_data_context = {}

Graceful Degradation:

• If TuShare unavailable → Return general educational response
• If cache fails → Attempt fresh fetch
• If all fail → Return mock response with explanation

Health Check Endpoint

@router.get("/api/ai/health")
async def health_check() -> Dict:
    fetcher = get_fundamental_fetcher()
    return {
        "status": "healthy",
        "fundamental_fetcher": "initialized" if fetcher else "not available",
        "tushare_available": bool(settings.tushare_token),
    }

Metrics Returned:

• Overall health status
• Data fetcher availability
• API key configuration status

Production Considerations

1. Rate Limiting: TuShare API has rate limits

   • Implement exponential backoff
   • Cache aggressively
   • Batch requests where possible

2. Data Freshness:

   • Quotes: 60-second cache
   • Fundamentals: 24-hour cache
   • Features: 5-minute cache

3. User Experience:

   • Provide instant responses for cached data
   • Show "fetching real-time data..." for misses
   • Timeout after 5 seconds with fallback

---

Summary: What Gets Built Next

Phase 2 Deliverables

1. sentiment_fetcher.py - Market sentiment data pipeline
2. announcement_crawler.py - Corporate announcements extraction
3. sentiment_analyzer.py - NLP-based sentiment scoring
4. market_context_builder.py - Context aggregation
5. Updated recommendation API with sentiment context
6. Updated AI assistant with market-aware responses
7. New database tables for announcements & sentiment
8. Phase 2 documentation & test suite

Effort: ~28 hours | Timeline: 4 weeks

Phase 3 Deliverables

1. knowledge_base.py - Vector database for research documents
2. cache_optimizer.py - Multi-level caching strategy
3. performance_monitor.py - System metrics & monitoring
4. batch_processor.py - Efficient batch operations
5. Database optimizations (indexes, partitioning, materialized views)
6. Kubernetes deployment manifests
7. Load testing suite & performance benchmarks
8. Production deployment guide
9. Phase 3 documentation & runbooks

Effort: ~34 hours | Timeline: 5 weeks

Total Project Timeline

Phase	Content	Duration	Status
1	Data pipeline + TuShare integration	Complete	✅ Done
2	Sentiment + Announcements	4 weeks	Pending
3	Performance + Knowledge base	5 weeks	Pending
Total	Full production system	13 weeks	30% complete