QuantDinger：下一代AI量化交易系统

0428

⏱️ 阅读时间：约10分钟

✨ 关注我们获取更多量化知识干货

项目介绍

QuantDinger是一个本地优先、隐私优先的自托管量化交易平台，专为追求数据主权和透明度的交易者、研究员和工程师设计。它不仅提供了完整的量化交易基础设施，还集成了先进的AI多智能体研究系统，帮助用户从海量市场数据中挖掘投资机会。

核心价值主张

QuantDinger解决了传统量化平台的几个痛点：

  • 数据主权：所有策略、交易日志、API密钥都存储在您的本地机器上，无厂商锁定
  • 透明审计：开源代码，可审查的交易基础设施
  • 工程导向：重视工程实现而非营销宣传
  • 完整工作流：涵盖数据获取、分析、回测和执行的全流程

技术特色

  • Apache 2.0开源许可证：商业友好的开源协议，允许自由修改和分发
  • Python原生可视化指标开发：使用标准Python编写指标（比PineScript更简单），并在内置K线图上直接运行
  • AI闭环优化：不仅仅是运行策略，AI会分析回测结果并建议参数调优（止损/止盈/MACD设置等），形成闭合优化循环
  • 全球市场访问：统一系统支持加密货币（实盘）、美股/港股（通过IBKR）、外汇（通过MT5）等
  • Docker现代化架构：4行命令部署，现代技术栈（Vue + Python）和干净的分层架构
图片[1]-QuantDinger：下一代AI量化交易系统-AI Express News

系统架构

整体架构

QuantDinger采用前后端分离的现代化架构：

┌─────────────────────────────┐
│      quantdinger_vue         │
│   (Vue 2 + Ant Design Vue)   │
└──────────────┬──────────────┘
               │  HTTP (/api/*)
               ▼
┌─────────────────────────────┐
│     backend_api_python       │
│   (Flask + strategy runtime) │
└──────────────┬──────────────┘
               │
               ├─ PostgreSQL (多用户支持)
               ├─ Redis (可选缓存)
               └─ 数据提供商 / LLMs / 交易所

后端架构

后端采用Python Flask框架构建，具有以下特点：

  1. 模块化设计：各个功能模块解耦，包括数据源、策略、回测、AI分析、交易执行等
  2. 插件化数据源：通过工厂模式支持多种市场数据源（加密货币、股票、外汇、期货）
  3. 多交易所集成：支持币安、OKX、Bitget、Bybit等10+交易所的直接API交易
  4. AI智能体系统：包含多个专业AI分析师，如市场分析师、基本面分析师、新闻分析师、情绪分析师、风险分析师等

前端架构

前端基于Vue 2构建，具备以下功能：

  1. 可视化策略编辑器：直观的界面设计，支持拖拽式指标配置
  2. K线图表：集成专业的金融图表库，支持多种技术指标显示
  3. 实时监控面板：实时监控市场动态、资产和策略状态
  4. 多语言支持：支持英语、简体中文、繁体中文、日语、韩语等多种语言

AI多智能体研究系统

QuantDinger的核心创新之一是其AI多智能体研究系统，该系统模拟了一个24/7运作的投资委员会

内存增强智能体

QuantDinger的AI智能体不是从零开始每次分析，而是利用本地内存存储和可选的反思/验证循环：

  • 本地RAG：经验检索注入智能体提示中（非模型微调）
  • 存储位置：PostgreSQL数据库（与主数据共享）或本地backend_api_python/data/memory/文件（隐私优先）
  • 检索排名公式：

    score = w_sim·sim + w_recency·recency + w_returns·returns_score

策略执行架构

  • 线程基础执行器：策略执行的独立线程池
  • 自动恢复：系统重启后恢复运行中的策略
  • 订单队列：后台工作者处理订单执行

快速入手

Docker部署（推荐）

这是最快获得QuantDinger运行的方式，支持PostgreSQL数据库和多用户功能。

1. 配置环境

创建项目根目录下的.env文件：

# Database Configuration
POSTGRES_USER=quantdinger
POSTGRES_PASSWORD=your_secure_password
POSTGRES_DB=quantdinger

# Admin Account (created on first startup)
ADMIN_USER=quantdinger
ADMIN_PASSWORD=123456

# Optional: AI Features
OPENROUTER_API_KEY=your_api_key

2. 启动服务

Linux / macOS

git clone https://github.com/brokermr810/QuantDinger.git && 
cd QuantDinger && 
cp backend_api_python/env.example backend_api_python/.env && 
docker-compose up -d --build

Windows (PowerShell)

git clone https://github.com/brokermr810/QuantDinger.git
cd QuantDinger
Copy-Item backend_api_pythonenv.example -Destination backend_api_python.env
docker-compose up -d --build

这将自动：

  • 启动PostgreSQL数据库（端口5432）
  • 初始化数据库架构
  • 启动后端API（端口5000）
  • 启动前端（端口8888）
  • 使用.env中的ADMIN_USER/ADMIN_PASSWORD创建管理员用户

3. 访问应用

  • 前端UI：http://localhost:8888
  • 后端API：http://localhost:5000
  • 默认账户：使用.env中的ADMIN_USER / ADMIN_PASSWORD（默认：quantdinger / 123456，生产环境中请更改）

本地开发部署

如果您想进行开发或自定义，可以使用本地部署方式：

前提条件

  • Python 3.10+ 推荐
  • Node.js 16+ 推荐
  • PostgreSQL 14+ 已安装并运行

1. 设置PostgreSQL

# 创建数据库和用户
sudo -u postgres psql
CREATE DATABASE quantdinger;
CREATE USER quantdinger WITH ENCRYPTED PASSWORD 'your_password';
GRANT ALL PRIVILEGES ON DATABASE quantdinger TO quantdinger;
q

# 初始化架构
psql -U quantdinger -d quantdinger -f backend_api_python/migrations/init.sql

2. 启动后端（Flask API）

cd backend_api_python
pip install -r requirements.txt
cp env.example .env   # Windows: copy env.example .env

编辑.env并设置：

DATABASE_URL=postgresql://quantdinger:your_password@localhost:5432/quantdinger
SECRET_KEY=your-secret-key
ADMIN_USER=quantdinger
ADMIN_PASSWORD=123456

然后启动：

python run.py

后端将在http://localhost:5000可用。

3. 启动前端（Vue UI）

cd quantdinger_vue
npm install
npm run serve

前端开发服务器运行在http://localhost:8000，并将/api/*代理到http://localhost:5000。

案例展示

核心策略代码示例

QuantDinger的策略编译器将策略配置编译为可执行的Python代码。以下是生成的策略代码示例：

# 生成的策略：我的SuperTrend策略
import pandas as pd
import numpy as np

# 检查信号的安全辅助函数
def get_val(arr, i, default=0):
    if i < 0 or i >= len(arr): return default
    return arr[i]

# ===========================
# 1. 参数
# ===========================
initial_position_pct = 0.1
leverage = 1
max_pyramiding = 0

# 加仓
add_position_pct = 0
add_threshold_pct = 0.0

# 风险管理
stop_loss_pct = 0.05
take_profit_activation = 0.02
trailing_callback = 0.01

# ===========================
# 2. 指标计算
# ===========================
# SuperTrend (14, 3.0)
period = 14
multiplier = 3.0
df['hl2'] = (df['high'] + df['low']) / 2
df['tr'] = np.maximum(df['high'] - df['low'], np.maximum(abs(df['high'] - df['close'].shift(1)), abs(df['low'] - df['close'].shift(1))))
df['atr'] = df['tr'].ewm(alpha=1/period, adjust=False).mean()
df['basic_upper'] = df['hl2'] + (multiplier * df['atr'])
df['basic_lower'] = df['hl2'] - (multiplier * df['atr'])

final_upper = [0.0] * len(df)
final_lower = [0.0] * len(df)
trend = [1] * len(df)
close_arr = df['close'].values
basic_upper = np.nan_to_num(df['basic_upper'].values)
basic_lower = np.nan_to_num(df['basic_lower'].values)

for i in range(1, len(df)):
    if basic_upper[i] < final_upper[i-1] or close_arr[i-1] > final_upper[i-1]:
        final_upper[i] = basic_upper[i]
    else:
        final_upper[i] = final_upper[i-1]
        
    if basic_lower[i] > final_lower[i-1] or close_arr[i-1] < final_lower[i-1]:
        final_lower[i] = basic_lower[i]
    else:
        final_lower[i] = final_lower[i-1]
        
    prev_trend = trend[i-1]
    if prev_trend == -1 and close_arr[i] > final_upper[i-1]:
        trend[i] = 1
    elif prev_trend == 1 and close_arr[i] < final_lower[i-1]:
        trend[i] = -1
    else:
        trend[i] = prev_trend

df['st_trend'] = trend
df['st_upper'] = final_upper
df['st_lower'] = final_lower

# ===========================
# 3. 入场信号逻辑
# ===========================
# 默认为False
df['raw_buy'] = False
df['raw_sell'] = False

df['raw_buy'] = (df['st_trend'] == 1) & (df['st_trend'].shift(1) == -1)
df['raw_sell'] = (df['st_trend'] == -1) & (df['st_trend'].shift(1) == 1)

AI智能体分析示例

以下是BullResearcher（看涨研究员）的部分代码，展示了如何构建看涨论证：

class BullResearcher(BaseAgent):
    """看涨研究员。"""
    
    def analyze(self, context: Dict[str, Any]) -> Dict[str, Any]:
        """构建看涨论证。"""
        market = context.get('market')
        symbol = context.get('symbol')
        language = context.get('language', 'zh-CN')
        model = context.get('model')
        
        # 输入
        market_report = context.get('market_report', {})
        fundamental_report = context.get('fundamental_report', {})
        news_report = context.get('news_report', {})
        sentiment_report = context.get('sentiment_report', {})
        
        # 记忆
        situation = f"{market}:{symbol} 看涨论证"
        memory_meta = {
            "market": market,
            "symbol": symbol,
            "timeframe": context.get("timeframe"),
            "features": context.get("memory_features") or {},
        }
        memories = self.get_memories(situation, n_matches=None, metadata=memory_meta)
        memory_prompt = self.format_memories_for_prompt(memories)
        
        lang_instruction = self._get_language_instruction(language)
        system_prompt = f"""您是一名看涨分析师，为投资决策构建看涨论证。您的任务是：
{lang_instruction}
1. 突出增长潜力、竞争优势和积极的市场指标。
2. 使用提供的研究和数据构建强有力的论证。
3. 有效解决/反驳看空观点。
4. 从历史经验中学习：{memory_prompt}
5. **信心评分**：评估您对看涨论证的信心（0-100）。现实一点。如果数据混合或薄弱，请降低信心。不要默认为75。

请按以下JSON格式返回：
{{
  "argument": "详细的看涨论证...",
  "key_points": ["要点1", "要点2", "要点3"],
  "confidence": 75
}}"""

        user_prompt = f"""基于以下分析报告，为{market}市场的{symbol}构建看涨论证：

**市场技术分析：**
{json.dumps(market_report.get('data', {}), ensure_ascii=False, indent=2) if market_report else '无数据'}

**基本面分析：**
{json.dumps(fundamental_report.get('data', {}), ensure_ascii=False, indent=2) if fundamental_report else '无数据'}

**新闻分析：**
{json.dumps(news_report.get('data', {}), ensure_ascii=False, indent=2) if news_report else '无数据'}

**情绪分析：**
{json.dumps(sentiment_report.get('data', {}), ensure_ascii=False, indent=2) if sentiment_report else '无数据'}

请构建一个强有力的看涨论证，强调增长潜力、竞争优势和积极指标。"""

        result = self.llm_service.safe_call_llm(
            system_prompt,
            user_prompt,
            {"argument": "", "key_points": [], "confidence": 50},
            model=model
        )
        
        return {
            "type": "bull",
            "data": result
        }

交易执行器示例

交易执行器负责实时执行交易策略，以下是其关键部分：

def _execute_indicator_df(
    self, indicator_code: str, df: pd.DataFrame, trading_config: Dict[str, Any], 
    initial_highest_price: float = 0.0,
    initial_position: int = 0,
    initial_avg_entry_price: float = 0.0,
    initial_position_count: int = 0,
    initial_last_add_price: float = 0.0
) -> tuple[Optional[pd.DataFrame], dict]:
    """执行指标代码，返回执行后的DataFrame和执行环境"""
    try:
        # 确保 DataFrame 的所有数值列都是 float64 类型
        df = df.copy()
        for col in ['open', 'high', 'low', 'close', 'volume']:
            if col in df.columns:
                if not pd.api.types.is_numeric_dtype(df[col]):
                    df[col] = pd.to_numeric(df[col], errors='coerce').astype('float64')
                else:
                    df[col] = df[col].astype('float64')
        
        # 删除包含 NaN 的行
        df = df.dropna()
        
        if len(df) == 0:
            logger.warning("DataFrame is empty; cannot execute indicator script")
            return None, {}
        
        # 初始化信号Series
        signals = pd.Series(0, index=df.index, dtype='float64')
        
        # 准备执行环境
        tc = dict(trading_config or {})
        cfg = self._build_cfg_from_trading_config(tc)
        local_vars = {
            'df': df,
            'open': df['open'].astype('float64'),
            'high': df['high'].astype('float64'),
            'low': df['low'].astype('float64'),
            'close': df['close'].astype('float64'),
            'volume': df['volume'].astype('float64'),
            'signals': signals,
            'np': np,
            'pd': pd,
            'trading_config': tc,
            'config': tc,  # 别名
            'cfg': cfg,    # 标准化嵌套配置
            'leverage': float(trading_config.get('leverage', 1)),
            'initial_capital': float(trading_config.get('initial_capital', 1000)),
            'commission': 0.001,
            'trade_direction': str(trading_config.get('trade_direction', 'long')),
            'initial_highest_price': float(initial_highest_price),
            'initial_position': int(initial_position),
            'initial_avg_entry_price': float(initial_avg_entry_price),
            'initial_position_count': int(initial_position_count),
            'initial_last_add_price': float(initial_last_add_price)
        }
        
        # 执行指标代码
        exec_env = local_vars.copy()
        exec_env['__builtins__'] = safe_builtins
        
        pre_import_code = "import numpy as npnimport pandas as pdn"
        exec(pre_import_code, exec_env)
        
        exec(indicator_code, exec_env)
        
        executed_df = exec_env.get('df', df)
        return executed_df, exec_env
        
    except Exception as e:
        logger.error(f"Failed to execute indicator script: {str(e)}")
        logger.error(traceback.format_exc())
        return None, {}

总结

QuantDinger提供了一个强大而灵活的解决方案，既保证了数据安全，又充分利用了人工智能的力量。

项目地址

https://github.com/brokermr810/QuantDinger

关注【魔方量化】，带你玩转开源量化交易！

 

<原文链接：https://mp.weixin.qq.com/s/wQabwClsfT3x0iP4KmqAkg

© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
AI 资讯
喜欢就支持一下吧
点赞8 分享
魔方量化实验室的头像-AI Express News
相关推荐
AntiGravity 完美兼容Claude Skills？！-AI Express News
AntiGravity 完美兼容Claude Skills？！
AntiGravity 完美兼容Claude Skills？！
1个月前 153
147 次失败后的神级 Prompt：Reddit 爆火的「Lyra」是什么？-AI Express News
147 次失败后的神级 Prompt：Reddit 爆火的「Lyra」是什么？
147 次失败后的神级 Prompt：Reddit 爆火的「Lyra」是什么？
3个月前 137
多地探索将 DeepSeek 嵌入纪检监察工作，甄别可疑数据-澎湃新闻-AI Express News
多地探索将 DeepSeek 嵌入纪检监察工作，甄别可疑数据-澎湃新闻
多地探索将 DeepSeek 嵌入纪检监察工作，甄别可疑数据-澎湃新闻
10个月前 111
470亿买下美国AI芯片龙头，67岁的孙正义押注下一个万亿英伟达-AI Express News
470亿买下美国AI芯片龙头，67岁的孙正义押注下一个万亿英伟达
470亿买下美国AI芯片龙头，67岁的孙正义押注下一个万亿英伟达
10个月前 110
Nano Banana｜生活照变高质量女士证件照教程（附提示词全文）-AI Express News
Nano Banana｜生活照变高质量女士证件照教程（附提示词全文）
Nano Banana｜生活照变高质量女士证件照教程（附提示词全文）
3个月前 85
AI 每日速递 20251017-AI Express News
AI 每日速递 20251017
AI 每日速递 20251017
3个月前 85
评论 抢沙发

请登录后发表评论

    暂无评论内容