🇨🇳 简体中文
🇺🇸 English
🇯🇵 日本語
Skip to the content.

开发指南

基于 FastAPI 和 React 的量化交易系统开发指南。

快速开始

# 启动所有服务
./scripts/startup.sh

# 仅启动后端
./scripts/startup.sh --backend-only

# 仅启动前端
./scripts/startup.sh --frontend-only

访问:

环境配置

必需环境变量

# Redis 配置
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD=  # 可选

# 环境类型
ENVIRONMENT=development

可选环境变量

# 长桥 API 配置,development 环境中用户无设置时这里兜底,便于测试
LONGPORT_APP_KEY=your_app_key
LONGPORT_APP_SECRET=your_app_secret
LONGPORT_ACCESS_TOKEN=your_access_token
LONGPORT_REGION=HK
LONGPORT_LANGUAGE=zh-CN

AUTHORIZATION=Bearer your-jwt-token-here

# 强制 Mock 模式(测试用)
FORCE_MOCK_MODE=true

配置步骤

# 1. 复制模板
cp .env.example .env

# 2. 编辑配置
vim .env

认证配置

默认用户

开发环境默认用户:

生产环境注意:

  1. 立即创建管理员账户
  2. 删除默认测试用户
  3. 定期检查账户安全

开发环境认证 Fallback

开发环境支持通过环境变量 AUTHORIZATION 进行免登录认证。

获取 JWT Token:

curl -X POST http://localhost:8000/api/v1/login \
  -H "Content-Type: application/json" \
  -d '{
    "username": "test_for_dev",
    "password": "test_only_for_dev",
    "captcha": "1234"
  }'

设置环境变量:

# 在 .env 文件中添加
AUTHORIZATION=Bearer your-jwt-token-here

认证优先级:

  1. 请求头 Authorization(最高)
  2. 环境变量 AUTHORIZATION
  3. 无认证(返回 401)

注意: 此功能仅在非生产环境生效。

开发规范

日期时间处理

前端规范

API 调用规范

// ✅ 正确:相对路径
const response = await fetch('/api/v1/users/me', {
  headers: { 'Authorization': `Bearer ${token}` }
});

// ❌ 错误:硬编码完整 URL
const response = await fetch('http://localhost:8000/api/v1/users/me', {
  headers: { 'Authorization': `Bearer ${token}` }
});

Vite 代理配置

// vite.config.ts
server: {
  port: 3000,
  proxy: {
    '/api': {
      target: 'http://127.0.0.1:8000',
      changeOrigin: true,
      secure: false
    }
  }
}

代码风格

文件管理

测试

代码检查

# 格式化
cd backend && source venv/bin/activate
black .
isort .

# 代码检查
flake8 .
mypy .

日志查看

抽象接口开发

核心原则

  1. 业务代码禁止直接使用 SDK
  2. 统一接口规范
  3. 透明切换券商
  4. 类型安全

示例

# ✅ 正确:通过抽象接口
class StockService:
    def __init__(self):
        self.data_source = data_source_factory.get_client(user_id)

    def get_stock_quotes(self, symbols):
        return self.data_source.get_quote(symbols)

# ❌ 错误:直接使用 SDK
from longport import QuoteContext
class StockService:
    def __init__(self):
        self.quote_context = QuoteContext(config)  # 禁止

添加新接口

1. 在抽象基类中声明:

class BaseDataSourceClient(ABC):
    @abstractmethod
    def get_new_data(self, param1: str, param2: int) -> dict:
        """获取新数据"""
        pass

2. 在所有实现类中实现:

class LongPortDataSourceClient(BaseDataSourceClient):
    def get_new_data(self, param1: str, param2: int) -> dict:
        return self._quote_context.new_data_method(param1, param2)

3. 业务层直接使用:

class BusinessService:
    def process_new_data(self, param1, param2):
        client = data_source_factory.get_client(user_id)
        return client.get_new_data(param1, param2)

数据转换处理

枚举类型转换:

def _get_enum_value(self, enum_name: str, value: str):
    """将字符串转换为 SDK 枚举"""
    enum_mappings = {
        "Market": {
            "US": longport.Market.US,
            "HK": longport.Market.HK,
            "CN": longport.Market.CN,
        }
    }

    if enum_name in enum_mappings and value in enum_mappings[enum_name]:
        return enum_mappings[enum_name][value]
    raise ValueError(f"无效的 {enum_name} 值: {value}")

日期格式处理:

def _parse_date(self, date_input):
    """统一日期格式处理"""
    from datetime import datetime, date

    if isinstance(date_input, date):
        return date_input
    elif isinstance(date_input, datetime):
        return date_input.date()
    elif isinstance(date_input, str):
        dt = datetime.fromisoformat(date_input.replace('Z', '+00:00'))
        return dt.date()
    else:
        raise ValueError(f"不支持的日期格式: {type(date_input)}")

添加新券商支持

1. 实现数据源客户端:

class NewBrokerDataSourceClient(BaseDataSourceClient):
    def __init__(self, config: DataSourceConfig):
        super().__init__(config)
        self._client = None

    def initialize(self) -> bool:
        try:
            self._client = NewBrokerSDK(
                api_key=self.config.config.get("api_key"),
                secret=self.config.config.get("secret")
            )
            return True
        except Exception as e:
            print(f"初始化失败: {e}")
            return False

    def get_quote(self, symbols: list) -> list:
        if not self._initialized:
            raise RuntimeError("客户端未初始化")
        result = self._client.get_realtime_quotes(symbols)
        return self._serialize_result(result)

2. 注册到工厂:

def create_client(config: DataSourceConfig) -> BaseDataSourceClient:
    if config.type == "longport":
        return LongPortDataSourceClient(config)
    elif config.type == "newbroker":
        return NewBrokerDataSourceClient(config)
    else:
        raise ValueError(f"不支持的数据源类型: {config.type}")

3. 更新配置模型:

class DataSourceType(str, Enum):
    LONGPORT = "longport"
    NEWBROKER = "newbroker"

注意事项

  1. 认证 Fallback 仅用于开发环境
  2. JWT Token 有 7 天有效期
  3. 不要在代码中硬编码 Token
  4. 测试后清理环境变量
  5. 使用 API 测试模块验证功能
  6. 确保多券商配置格式正确