Python自动化炒股：使用FastAPI和Docker Compose部署股票数据服务的最佳实践

在当今的金融市场中，自动化交易已经成为一个重要的趋势。Python因其强大的数据处理能力和丰富的库支持，成为了自动化炒股的首选语言。本文将介绍如何使用FastAPI和Docker Compose来部署一个股票数据服务，以支持自动化交易系统。

引言

自动化炒股系统通常需要实时或近实时的股票数据。FastAPI是一个现代、快速（高性能）的Web框架，用于构建APIs，而Docker Compose则用于定义和运行多容器Docker应用程序。结合两者，我们可以轻松地部署和管理股票数据服务。

环境准备

在开始之前，请确保你的环境中安装了以下软件：

• Python 3.8 或更高版本
• Docker
• Docker Compose

1. 创建FastAPI应用

首先，我们需要创建一个FastAPI应用来提供股票数据服务。

1.1 安装FastAPI

在你的Python环境中安装FastAPI：

pip install fastapi uvicorn

1.2 创建FastAPI应用

创建一个名为mAIn.py的文件，并添加以下代码：

from fastapi import FastAPI
from typing import List

app = FastAPI()

@app.get("/stock/{symbol}", response_model=List[dict])
async def read_stock(symbol: str):
    # 这里应该是调用股票数据API的代码，为了示例，我们使用静态数据
    stock_data = {
        "AAPL": {"price": 150, "volume": 10000},
        "GOOGL": {"price": 2800, "volume": 5000},
    }
    return [stock_data.get(symbol, {"error": "Stock symbol not found"})]

这段代码定义了一个简单的API，它接受一个股票符号作为参数，并返回该股票的价格和交易量。

2. 使用Docker Compose部署

2.1 创建Dockerfile

在项目根目录下创建一个Dockerfile文件，用于构建FastAPI应用的Docker镜像：

# 使用官方Python镜像
FROM python:3.8-slim

# 设置工作目录
WORKDIR /app

# 复制项目文件到工作目录
COPY . .

# 安装依赖
RUN pip install --no-cache-dir -r requirements.txt

# 运行FastAPI应用
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

2.2 创建requirements.txt

在项目根目录下创建一个requirements.txt文件，列出所有依赖：

fastapi
uvicorn

2.3 创建docker-compose.yml

创建一个docker-compose.yml文件，定义服务：

version: '3.8'
services:
  web:
    build: .
    ports:
      - "8000:8000"
    volumes:
      - .:/app
    command: uvicorn main:app --host 0.0.0.0 --port 8000

这个配置文件定义了一个名为web的服务，它使用我们之前创建的Docker镜像，并映射端口8000。

3. 启动服务

在项目根目录下运行以下命令来启动服务：

docker-compose up --build

这个命令会构建Docker镜像，并启动服务。你可以通过访问http://localhost:8000/stock/AAPL来测试API。

4. 实时股票数据集成

为了使系统更加实用，我们需要集成实时股票数据。这通常涉及到调用第三方API，如Alpha Vantage、IEX Cloud等。

4.1 安装第三方库

安装requests库来发送HTTP请求：

pip install requests

4.2 修改FastAPI应用

修改main.py文件，以集成实时股票数据：

from fastapi import FastAPI
import requests

app = FastAPI()

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://www.alphavantage.co/query"

@app.get("/stock/{symbol}", response_model=List[dict])
async def read_stock(symbol: str):
    params = {
        "function": "TIME_SERIES_INTRADAY",
        "symbol": symbol,
        "interval": "5min",
        "apikey": API_KEY
    }
    response = requests.get(BASE_URL, params=params)
    data = response.json()
    return [{"price": data["Time Series (5min)"]["2023-03