Jesse: 内置30+技术指标的高级Python加密货币交易框架 —— 2026年完整部署指南

python3 -m venv jesse-env
source jesse-env/bin/activate

# 安装Jesse
pip install jesse==1.7.2

# 创建项目目录
mkdir my-trading-bot && cd my-trading-bot

# 初始化Jesse（创建config、routes、strategies文件夹）
jesse init

# config.py — 数据库配置
DATABASES = {
    'default': {
        'driver': 'postgres',
        'host': 'localhost',
        'port': 5432,
        'dbname': 'jesse_db',
        'user': 'jesse_user',
        'password': 'your_secure_password'
    }
}

DATABASES = {
    'default': {
        'driver': 'sqlite',
        'path': 'storage/jesse.db'
    }
}

# routes.py — 定义交易对
from jesse.enums import timeframes

routes = [
    {'exchange': 'Binance', 'symbol': 'BTC-USDT', 'timeframe': '1h', 'strategy': 'SimpleMA'},
    {'exchange': 'Binance', 'symbol': 'ETH-USDT', 'timeframe': '1h', 'strategy': 'SimpleMA'},
]

extra_candles = [
    {'exchange': 'Binance', 'symbol': 'BTC-USDT', 'timeframe': '4h'},
]

# 从Binance下载1年的1小时BTC-USDT K线数据
jesse import-candles Binance BTC-USDT 2025-01-01

# strategies/SimpleMA/__init__.py
from jesse.strategies import Strategy
import jesse.indicators as ta

class SimpleMA(Strategy):
    def __init__(self):
        super().__init__()
        self.period = 20

    def should_long(self) -> bool:
        # 当价格上穿20周期SMA时做多
        sma = ta.sma(self.candles, self.period)
        return self.close > sma and self.close[-2] <= sma[-2]

    def should_short(self) -> bool:
        return False  # 本简单示例不做空

    def go_long(self):
        qty = self.capital / self.close
        self.buy = qty, self.close

    def go_short(self):
        pass

    def update_position(self):
        # 当价格跌破SMA时平仓
        sma = ta.sma(self.candles, self.period)
        if self.close < sma:
            self.liquidate()

# 运行routes中定义周期的回测
jesse backtest 2025-01-01 2025-12-31

# 使用NumPy的自定义指标
import numpy as np
import jesse.indicators as ta

def custom_zscore(candles, period=20):
    closes = np.array([c[2] for c in candles[-period:]])
    return (closes[-1] - closes.mean()) / closes.std()

class ZScoreStrategy(Strategy):
    def should_long(self):
        z = custom_zscore(self.candles, 20)
        return z < -2.0  # 价格低于均值2个标准差时买入

# 使用sklearn的ML增强策略
from sklearn.ensemble import RandomForestClassifier
import numpy as np

class MLStrategy(Strategy):
    def __init__(self):
        super().__init__()
        self.model = RandomForestClassifier(n_estimators=100)
        self.features = []
        self.labels = []

    def should_long(self):
        rsi = ta.rsi(self.candles, 14)
        sma20 = ta.sma(self.candles, 20)
        sma50 = ta.sma(self.candles, 50)
        atr = ta.atr(self.candles, 14)

        features = [rsi, sma20/sma50, atr/self.close]
        prediction = self.model.predict([features])
        return prediction[0] == 1

# config.py — Telegram通知设置
NOTIFICATIONS = {
    'enabled': True,
    'provider': 'telegram',
    'telegram_bot_token': 'YOUR_BOT_TOKEN',
    'telegram_chat_id': 'YOUR_CHAT_ID',
    'events': ['order_executed', 'trade_completed', 'error']
}

# Jesse部署的Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
CMD ["jesse", "run"]

# docker-compose.yml
version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: jesse_db
      POSTGRES_USER: jesse_user
      POSTGRES_PASSWORD: your_password
    volumes:
      - pgdata:/var/lib/postgresql/data

  jesse:
    build: .
    depends_on:
      - postgres
    environment:
      DATABASE_URL: postgres://jesse_user:your_password@postgres:5432/jesse_db
    volumes:
      - ./strategies:/app/strategies
      - ./config.py:/app/config.py
      - ./routes.py:/app/routes.py

volumes:
  pgdata:

# metrics.py — 导出Prometheus指标
from prometheus_client import Counter, Gauge, start_http_server

trades_total = Counter('jesse_trades_total', 'Total trades executed')
position_size = Gauge('jesse_position_size', 'Current position size')
pnl_current = Gauge('jesse_pnl_percent', 'Current P&L percentage')

# 在9090端口启动指标服务器
start_http_server(9090)

# config.py — 风险管理设置
RISK_MANAGEMENT = {
    'max_risk_per_trade': 0.02,      # 单笔交易最大风险2%
    'max_drawdown_stop': 0.15,       # 回撤15%时停止交易
    'daily_loss_limit': 0.05,        # 日亏损限制5%
    'position_size_limit': 0.25,     # 单个仓位最大25%
}

# 多时间周期策略示例
class MultiTFStrategy(Strategy):
    def prepare(self):
        # 获取4小时K线用于趋势判断
        self.h4_candles = self.get_candles('Binance', 'BTC-USDT', '4h')

    def should_long(self):
        h4_sma50 = ta.sma(self.h4_candles, 50)
        h1_sma20 = ta.sma(self.candles, 20)

        # 仅在4小时趋势向上且1小时显示动能时做多
        return self.close_4h > h4_sma50 and self.close > h1_sma20

# 高级出场逻辑
class RiskManagedStrategy(Strategy):
    def go_long(self):
        entry = self.close
        stop_loss = entry * 0.97       # 3%止损
        take_profit = entry * 1.06     # 6%目标
        qty = (self.capital * 0.02) / (entry - stop_loss)

        self.buy = qty, entry
        self.stop_loss = qty, stop_loss
        self.take_profit = qty, take_profit

# 在模拟交易模式下运行（使用实时数据模拟订单）
jesse run --paper

# 实时监控日志
tail -f storage/logs/live-trading.log

# 每日备份定时任务
0 2 * * * pg_dump jesse_db | gzip > /backups/jesse_$(date +\%F).sql.gz

# config.py — 安全的API密钥处理
import os

EXCHANGES = {
    'Binance': {
        'api_key': os.environ['BINANCE_API_KEY'],
        'api_secret': os.environ['BINANCE_API_SECRET'],
        'sandbox': False
    }
}