MomentumLab/app/docs/06-后端实现.md

# A股智投分析平台 - 后端实现文档

## 一、技术选型

### 1.1 推荐方案: Node.js + Express

```
技术栈:
- Node.js 20.x LTS
- Express 4.x
- TypeScript 5.x
- Prisma ORM
- Socket.io (WebSocket)
- Redis (缓存)
- MySQL 8.0 (数据库)
```

### 1.2 备选方案: Python + FastAPI

```
技术栈:
- Python 3.11
- FastAPI
- SQLAlchemy
- WebSockets
- Celery (定时任务)
- Redis
- PostgreSQL
```

---

## 二、项目结构

```
backend/
├── src/
│   ├── config/              # 配置文件
│   │   ├── database.ts      # 数据库配置
│   │   ├── redis.ts         # Redis配置
│   │   └── constants.ts     # 常量定义
│   │
│   ├── controllers/         # 控制器层
│   │   ├── marketController.ts
│   │   ├── sectorController.ts
│   │   ├── stockController.ts
│   │   └── userController.ts
│   │
│   ├── services/            # 业务逻辑层
│   │   ├── marketService.ts
│   │   ├── sectorService.ts
│   │   ├── stockService.ts
│   │   ├── dataSyncService.ts
│   │   └── calculationService.ts
│   │
│   ├── models/              # 数据模型层
│   │   ├── Stock.ts
│   │   ├── Sector.ts
│   │   ├── KLine.ts
│   │   └── User.ts
│   │
│   ├── routes/              # 路由定义
│   │   ├── marketRoutes.ts
│   │   ├── sectorRoutes.ts
│   │   ├── stockRoutes.ts
│   │   └── userRoutes.ts
│   │
│   ├── middleware/          # 中间件
│   │   ├── auth.ts          # 认证中间件
│   │   ├── errorHandler.ts  # 错误处理
│   │   ├── rateLimiter.ts   # 限流
│   │   └── logger.ts        # 日志
│   │
│   ├── utils/               # 工具函数
│   │   ├── logger.ts
│   │   ├── validator.ts
│   │   ├── formatter.ts
│   │   └── maCalculator.ts  # 均线计算
│   │
│   ├── websocket/           # WebSocket服务
│   │   └── stockSocket.ts
│   │
│   ├── jobs/                # 定时任务
│   │   ├── syncMarketData.ts
│   │   ├── calculateMomentum.ts
│   │   └── updateRankings.ts
│   │
│   ├── types/               # 类型定义
│   │   └── index.ts
│   │
│   └── app.ts               # 应用入口
│
├── prisma/                  # Prisma ORM
│   └── schema.prisma
│
├── tests/                   # 测试文件
│   ├── unit/
│   └── integration/
│
├── scripts/                 # 脚本文件
│   └── init-db.ts
│
├── .env                     # 环境变量
├── .env.example             # 环境变量示例
├── Dockerfile
├── docker-compose.yml
├── package.json
├── tsconfig.json
└── README.md
```

---

## 三、核心服务实现

### 3.1 市场数据服务

**文件**: `src/services/marketService.ts`

```typescript
import { PrismaClient } from '@prisma/client';
import Redis from 'ioredis';

const prisma = new PrismaClient();
const redis = new Redis(process.env.REDIS_URL);

export class MarketService {
    // 获取市场指数
    async getMarketIndices() {
        const cacheKey = 'market:indices';
        const cached = await redis.get(cacheKey);
        
        if (cached) {
            return JSON.parse(cached);
        }
        
        const indices = await prisma.marketIndex.findMany({
            orderBy: { sortOrder: 'asc' }
        });
        
        await redis.setex(cacheKey, 60, JSON.stringify(indices));
        return indices;
    }
    
    // 获取涨跌家数统计
    async getUpDownStats() {
        const cacheKey = 'market:updown:stats';
        const cached = await redis.get(cacheKey);
        
        if (cached) {
            return JSON.parse(cached);
        }
        
        const stats = await prisma.stockQuote.groupBy({
            by: ['changePercent'],
            _count: { code: true },
            where: {
                quoteTime: {
                    gte: new Date(Date.now() - 5 * 60 * 1000) // 5分钟内
                }
            }
        });
        
        const result = {
            up: stats.filter(s => s.changePercent > 0).reduce((a, b) => a + b._count.code, 0),
            down: stats.filter(s => s.changePercent < 0).reduce((a, b) => a + b._count.code, 0),
            flat: stats.filter(s => s.changePercent === 0).reduce((a, b) => a + b._count.code, 0)
        };
        
        await redis.setex(cacheKey, 60, JSON.stringify(result));
        return result;
    }
    
    // 获取涨跌幅分布
    async getPriceDistribution() {
        const cacheKey = 'market:price:distribution';
        const cached = await redis.get(cacheKey);
        
        if (cached) {
            return JSON.parse(cached);
        }
        
        const ranges = [
            { range: '<-7%', min: -100, max: -7 },
            { range: '-7~-5%', min: -7, max: -5 },
            { range: '-5~-3%', min: -5, max: -3 },
            { range: '-3~0%', min: -3, max: 0 },
            { range: '0~3%', min: 0, max: 3 },
            { range: '3~5%', min: 3, max: 5 },
            { range: '5~7%', min: 5, max: 7 },
            { range: '>7%', min: 7, max: 100 }
        ];
        
        const distribution = await Promise.all(
            ranges.map(async r => {
                const count = await prisma.stockQuote.count({
                    where: {
                        changePercent: {
                            gte: r.min,
                            lt: r.max
                        },
                        quoteTime: {
                            gte: new Date(Date.now() - 5 * 60 * 1000)
                        }
                    }
                });
                return { ...r, count };
            })
        );
        
        await redis.setex(cacheKey, 60, JSON.stringify(distribution));
        return distribution;
    }
}
```

### 3.2 版块数据服务

**文件**: `src/services/sectorService.ts`

```typescript
export class SectorService {
    // 获取版块列表（带动量排名）
    async getSectorsWithMomentum() {
        const cacheKey = 'sectors:momentum';
        const cached = await redis.get(cacheKey);
        
        if (cached) {
            return JSON.parse(cached);
        }
        
        const sectors = await prisma.sector.findMany({
            include: {
                quotes: {
                    orderBy: { quoteTime: 'desc' },
                    take: 1
                }
            }
        });
        
        // 计算动量分数和排名
        const sectorsWithMomentum = sectors.map(sector => {
            const latestQuote = sector.quotes[0];
            const momentumScore = this.calculateMomentumScore(sector);
            
            return {
                ...sector,
                changePercent: latestQuote?.changePercent || 0,
                momentumScore,
                rank: 0 // 稍后计算
            };
        });
        
        // 按动量分数排序
        sectorsWithMomentum.sort((a, b) => b.momentumScore - a.momentumScore);
        
        // 分配排名
        sectorsWithMomentum.forEach((sector, index) => {
            sector.rank = index + 1;
        });
        
        await redis.setex(cacheKey, 60, JSON.stringify(sectorsWithMomentum));
        return sectorsWithMomentum;
    }
    
    // 计算动量分数
    private calculateMomentumScore(sector: any): number {
        // 基于涨跌幅、成交量、趋势等因素计算
        const latestQuote = sector.quotes[0];
        if (!latestQuote) return 50;
        
        let score = 50;
        
        // 涨跌幅贡献 (0-30分)
        score += Math.min(Math.max(latestQuote.changePercent * 3, -15), 15);
        
        // 成交量贡献 (0-20分)
        const volumeRatio = latestQuote.volume / (latestQuote.avgVolume || latestQuote.volume);
        score += Math.min((volumeRatio - 1) * 10, 20);
        
        return Math.min(Math.max(score, 0), 100);
    }
    
    // 获取版块历史排名
    async getSectorRankHistory(sectorCode: string, days: number = 30) {
        const cacheKey = `sector:${sectorCode}:rank:history:${days}`;
        const cached = await redis.get(cacheKey);
        
        if (cached) {
            return JSON.parse(cached);
        }
        
        const history = await prisma.sectorQuote.findMany({
            where: {
                sectorCode,
                quoteTime: {
                    gte: new Date(Date.now() - days * 24 * 60 * 60 * 1000)
                }
            },
            orderBy: { quoteTime: 'asc' },
            select: {
                quoteTime: true,
                rank: true,
                momentumScore: true
            }
        });
        
        const result = history.map(h => ({
            date: h.quoteTime.toISOString().split('T')[0],
            rank: h.rank,
            momentumScore: h.momentumScore
        }));
        
        await redis.setex(cacheKey, 300, JSON.stringify(result));
        return result;
    }
    
    // 获取版块内动量股票
    async getSectorMomentumStocks(sectorCode: string) {
        const stocks = await prisma.stock.findMany({
            where: { sectorCode },
            include: {
                quotes: {
                    orderBy: { quoteTime: 'desc' },
                    take: 1
                }
            }
        });
        
        return stocks
            .map(stock => ({
                ...stock,
                ...stock.quotes[0],
                momentumScore: this.calculateStockMomentum(stock)
            }))
            .sort((a, b) => b.momentumScore - a.momentumScore);
    }
    
    private calculateStockMomentum(stock: any): number {
        const quote = stock.quotes[0];
        if (!quote) return 50;
        
        let score = 50;
        score += Math.min(Math.max(quote.changePercent * 4, -20), 20);
        
        const volumeRatio = quote.volume / (quote.avgVolume || quote.volume);
        score += Math.min((volumeRatio - 1) * 15, 25);
        
        return Math.min(Math.max(score, 0), 100);
    }
}
```

### 3.3 股票数据服务

**文件**: `src/services/stockService.ts`

```typescript
export class StockService {
    // 搜索股票
    async searchStocks(keyword: string) {
        const stocks = await prisma.stock.findMany({
            where: {
                OR: [
                    { name: { contains: keyword } },
                    { code: { contains: keyword } }
                ]
            },
            take: 10,
            include: {
                quotes: {
                    orderBy: { quoteTime: 'desc' },
                    take: 1
                }
            }
        });
        
        return stocks.map(s => ({
            ...s,
            ...s.quotes[0]
        }));
    }
    
    // 获取个股详情
    async getStockDetail(code: string) {
        const cacheKey = `stock:${code}:detail`;
        const cached = await redis.get(cacheKey);
        
        if (cached) {
            return JSON.parse(cached);
        }
        
        const stock = await prisma.stock.findUnique({
            where: { code },
            include: {
                quotes: {
                    orderBy: { quoteTime: 'desc' },
                    take: 1
                }
            }
        });
        
        if (!stock) return null;
        
        const klines = await prisma.kLineData.findMany({
            where: { stockCode: code, period: 'day' },
            orderBy: { date: 'desc' },
            take: 60
        });
        
        // 计算技术指标
        const indicators = this.calculateIndicators(klines);
        
        const result = {
            ...stock,
            ...stock.quotes[0],
            ...indicators
        };
        
        await redis.setex(cacheKey, 60, JSON.stringify(result));
        return result;
    }
    
    // 获取K线数据
    async getKLineData(code: string, period: string = 'day', days: number = 60) {
        const cacheKey = `stock:${code}:kline:${period}:${days}`;
        const cached = await redis.get(cacheKey);
        
        if (cached) {
            return JSON.parse(cached);
        }
        
        const klines = await prisma.kLineData.findMany({
            where: {
                stockCode: code,
                period
            },
            orderBy: { date: 'desc' },
            take: days
        });
        
        // 计算均线
        const klinesWithMA = this.calculateMA(klines.reverse());
        
        await redis.setex(cacheKey, 300, JSON.stringify(klinesWithMA));
        return klinesWithMA;
    }
    
    // 计算均线
    private calculateMA(klines: any[]) {
        const periods = [5, 10, 20, 30, 60];
        
        return klines.map((kline, index) => {
            const ma: Record<string, number> = {};
            
            for (const period of periods) {
                if (index >= period - 1) {
                    const sum = klines
                        .slice(index - period + 1, index + 1)
                        .reduce((acc, k) => acc + k.close, 0);
                    ma[`ma${period}`] = Number((sum / period).toFixed(2));
                }
            }
            
            return { ...kline, ...ma };
        });
    }
    
    // 计算技术指标
    private calculateIndicators(klines: any[]) {
        return {
            macd: this.calculateMACD(klines),
            kdj: this.calculateKDJ(klines),
            rsi: this.calculateRSI(klines)
        };
    }
    
    // MACD计算
    private calculateMACD(klines: any[]) {
        const closes = klines.map(k => k.close).reverse();
        const ema12 = this.EMA(closes, 12);
        const ema26 = this.EMA(closes, 26);
        const dif = ema12.map((v, i) => v - ema26[i]);
        const dea = this.EMA(dif, 9);
        const macd = dif.map((v, i) => (v - dea[i]) * 2);
        
        return {
            dif: Number(dif[dif.length - 1].toFixed(3)),
            dea: Number(dea[dea.length - 1].toFixed(3)),
            macd: Number(macd[macd.length - 1].toFixed(3))
        };
    }
    
    // KDJ计算
    private calculateKDJ(klines: any[], n: number = 9) {
        // KDJ计算逻辑
        // ...
        return { k: 75.2, d: 68.5, j: 88.6 };
    }
    
    // RSI计算
    private calculateRSI(klines: any[]) {
        // RSI计算逻辑
        // ...
        return { rsi6: 72.5, rsi12: 68.3, rsi24: 65.1 };
    }
    
    // EMA计算
    private EMA(data: number[], n: number): number[] {
        const k = 2 / (n + 1);
        const ema: number[] = [data[0]];
        
        for (let i = 1; i < data.length; i++) {
            ema.push(data[i] * k + ema[i - 1] * (1 - k));
        }
        
        return ema;
    }
}
```

### 3.4 数据同步服务

**文件**: `src/services/dataSyncService.ts`

```typescript
import axios from 'axios';

export class DataSyncService {
    private akshareBaseUrl = 'http://localhost:8000'; // AKShare服务地址
    
    // 同步实时行情
    async syncRealTimeQuotes() {
        try {
            // 从AKShare获取实时行情
            const response = await axios.get(`${this.akshareBaseUrl}/stock_zh_a_spot`);
            const quotes = response.data;
            
            // 批量插入数据库
            await prisma.$transaction(
                quotes.map((quote: any) =>
                    prisma.stockQuote.create({
                        data: {
                            stockCode: quote.code,
                            price: quote.price,
                            open: quote.open,
                            high: quote.high,
                            low: quote.low,
                            preClose: quote.pre_close,
                            volume: quote.volume,
                            turnover: quote.turnover,
                            changePercent: quote.change_percent,
                            quoteTime: new Date()
                        }
                    })
                )
            );
            
            console.log(`Synced ${quotes.length} quotes`);
        } catch (error) {
            console.error('Sync quotes failed:', error);
        }
    }
    
    // 同步K线数据
    async syncKLineData(stockCode: string, period: string = 'day') {
        try {
            const response = await axios.get(
                `${this.akshareBaseUrl}/stock_zh_a_hist`,
                {
                    params: {
                        symbol: stockCode,
                        period: period === 'day' ? 'daily' : period,
                        start_date: '20230101',
                        end_date: new Date().toISOString().split('T')[0].replace(/-/g, '')
                    }
                }
            );
            
            const klines = response.data;
            
            await prisma.$transaction(
                klines.map((k: any) =>
                    prisma.kLineData.upsert({
                        where: {
                            stockCode_period_date: {
                                stockCode: stockCode,
                                period: period,
                                date: new Date(k.date)
                            }
                        },
                        update: {
                            open: k.open,
                            high: k.high,
                            low: k.low,
                            close: k.close,
                            volume: k.volume
                        },
                        create: {
                            stockCode: stockCode,
                            period: period,
                            date: new Date(k.date),
                            open: k.open,
                            high: k.high,
                            low: k.low,
                            close: k.close,
                            volume: k.volume
                        }
                    })
                )
            );
            
            console.log(`Synced ${klines.length} klines for ${stockCode}`);
        } catch (error) {
            console.error(`Sync kline failed for ${stockCode}:`, error);
        }
    }
}
```

---

## 四、WebSocket 实现

**文件**: `src/websocket/stockSocket.ts`

```typescript
import { Server } from 'socket.io';

export class StockSocket {
    private io: Server;
    
    constructor(server: any) {
        this.io = new Server(server, {
            cors: {
                origin: '*',
                methods: ['GET', 'POST']
            }
        });
        
        this.setupHandlers();
    }
    
    private setupHandlers() {
        this.io.on('connection', (socket) => {
            console.log('Client connected:', socket.id);
            
            // 订阅股票
            socket.on('subscribe', (channels: string[]) => {
                channels.forEach(channel => {
                    socket.join(channel);
                    console.log(`Client ${socket.id} subscribed to ${channel}`);
                });
            });
            
            // 取消订阅
            socket.on('unsubscribe', (channels: string[]) => {
                channels.forEach(channel => {
                    socket.leave(channel);
                    console.log(`Client ${socket.id} unsubscribed from ${channel}`);
                });
            });
            
            socket.on('disconnect', () => {
                console.log('Client disconnected:', socket.id);
            });
        });
    }
    
    // 推送股票行情
    broadcastStockQuote(stockCode: string, data: any) {
        this.io.to(`stock:${stockCode}`).emit('quote', {
            channel: `stock:${stockCode}`,
            type: 'quote',
            data
        });
    }
    
    // 推送版块行情
    broadcastSectorQuote(sectorCode: string, data: any) {
        this.io.to(`sector:${sectorCode}`).emit('quote', {
            channel: `sector:${sectorCode}`,
            type: 'quote',
            data
        });
    }
}
```

---

## 五、定时任务

**文件**: `src/jobs/syncMarketData.ts`

```typescript
import cron from 'node-cron';
import { DataSyncService } from '../services/dataSyncService';

const dataSyncService = new DataSyncService();

// 每3秒同步实时行情（交易时间）
cron.schedule('*/3 * * * * *', async () => {
    const now = new Date();
    const hour = now.getHours();
    const minute = now.getMinutes();
    
    // 交易时间: 9:30-11:30, 13:00-15:00
    const isTradingTime = (
        (hour === 9 && minute >= 30) ||
        (hour === 10) ||
        (hour === 11 && minute <= 30) ||
        (hour === 13) ||
        (hour === 14)
    );
    
    if (isTradingTime) {
        await dataSyncService.syncRealTimeQuotes();
    }
});

// 每小时同步K线数据
cron.schedule('0 * * * *', async () => {
    const stocks = await prisma.stock.findMany();
    for (const stock of stocks) {
        await dataSyncService.syncKLineData(stock.code);
    }
});

// 每日收盘后计算版块排名
cron.schedule('0 15 * * 1-5', async () => {
    await sectorService.calculateAndUpdateRankings();
});
```

---

## 六、环境变量

**文件**: `.env`

```env
# 服务器配置
PORT=3000
NODE_ENV=production

# 数据库配置
DATABASE_URL=mysql://user:password@localhost:3306/aguzhitou

# Redis配置
REDIS_URL=redis://localhost:6379

# JWT配置
JWT_SECRET=your-secret-key
JWT_EXPIRES_IN=7d

# AKShare配置
AKSHARE_URL=http://localhost:8000

# 日志配置
LOG_LEVEL=info
```

---

## 七、部署脚本

**文件**: `docker-compose.yml`

```yaml
version: '3.8'

services:
  app:
    build: .
    ports:
      - "3000:3000"
    environment:
      - DATABASE_URL=mysql://root:rootpass@mysql:3306/aguzhitou
      - REDIS_URL=redis://redis:6379
      - JWT_SECRET=${JWT_SECRET}
    depends_on:
      - mysql
      - redis
    restart: always

  mysql:
    image: mysql:8.0
    environment:
      - MYSQL_ROOT_PASSWORD=rootpass
      - MYSQL_DATABASE=aguzhitou
    volumes:
      - mysql_data:/var/lib/mysql
    ports:
      - "3306:3306"
    restart: always

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
    ports:
      - "6379:6379"
    restart: always

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./ssl:/etc/nginx/ssl
    depends_on:
      - app
    restart: always

volumes:
  mysql_data:
  redis_data:
```

---

## 八、测试

```typescript
// 示例测试
describe('StockService', () => {
    let stockService: StockService;
    
    beforeEach(() => {
        stockService = new StockService();
    });
    
    it('should calculate MA correctly', () => {
        const klines = [
            { close: 10 }, { close: 11 }, { close: 12 },
            { close: 13 }, { close: 14 }, { close: 15 }
        ];
        
        const result = stockService['calculateMA'](klines);
        
        expect(result[5].ma5).toBe(13); // (11+12+13+14+15)/5 = 13
    });
});
```

---

## 九、待实现清单

- [ ] 数据库表创建
- [ ] Prisma schema 定义
- [ ] API 路由实现
- [ ] WebSocket 服务
- [ ] 定时任务配置
- [ ] 单元测试
- [ ] Docker 部署
- [ ] CI/CD 配置
- [ ] 监控告警
- [ ] 日志收集