gavindiaz/DinQuant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v3.0.1

Browse Source 
Signed-off-by: Dinger <quantdinger@gmail.com>
This commit is contained in:
Dinger
2026-04-06 01:39:25 +08:00
parent 15c901364b
commit 3ca291a346
214 changed files with 2771 additions and 8535 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend_api_python/app/routes/backtest.py
+68 -143
View File
@@ -265,44 +265,27 @@ def run_backtest():
'message': '使用标准K线回测'
}
# Persist backtest run for AI optimization / history
run_id = None
try:
with get_db_connection() as db:
cur = db.cursor()
cur.execute(
"""
INSERT INTO qd_backtest_runs
(user_id, indicator_id, market, symbol, timeframe, start_date, end_date,
initial_capital, commission, slippage, leverage, trade_direction,
strategy_config, status, error_message, result_json, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
""",
(
user_id,
int(indicator_id) if indicator_id is not None else None,
market,
symbol,
timeframe,
start_date_str,
end_date_str,
initial_capital,
commission,
slippage,
leverage,
trade_direction,
json.dumps(strategy_config or {}, ensure_ascii=False),
'success',
'',
json.dumps(result or {}, ensure_ascii=False)
)
)
run_id = cur.lastrowid
db.commit()
cur.close()
except Exception:
# Do not break the main backtest response if persistence fails.
logger.warning("Failed to persist backtest run", exc_info=True)
run_id = backtest_service.persist_run(
user_id=user_id,
indicator_id=int(indicator_id) if indicator_id is not None else None,
run_type='indicator',
market=market,
symbol=symbol,
timeframe=timeframe,
start_date_str=start_date_str,
end_date_str=end_date_str,
initial_capital=initial_capital,
commission=commission,
slippage=slippage,
leverage=leverage,
trade_direction=trade_direction,
strategy_config=strategy_config,
config_snapshot={'indicatorId': int(indicator_id) if indicator_id is not None else None},
status='success',
error_message='',
result=result,
code=indicator_code,
)
return jsonify({
'code': 1,
@@ -323,42 +306,31 @@ def run_backtest():
except Exception as e:
logger.error(f"Backtest failed: {str(e)}")
logger.error(traceback.format_exc())
# Best-effort persist failed run (if we have enough context)
try:
data = data if isinstance(data, dict) else {}
user_id = g.user_id
indicator_id = data.get('indicatorId')
with get_db_connection() as db:
cur = db.cursor()
cur.execute(
"""
INSERT INTO qd_backtest_runs
(user_id, indicator_id, market, symbol, timeframe, start_date, end_date,
initial_capital, commission, slippage, leverage, trade_direction,
strategy_config, status, error_message, result_json, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())
""",
(
user_id,
int(indicator_id) if indicator_id is not None else None,
str(data.get('market', '') or ''),
str(data.get('symbol', '') or ''),
str(data.get('timeframe', '') or ''),
str(data.get('startDate', '') or ''),
str(data.get('endDate', '') or ''),
float(data.get('initialCapital', 0) or 0),
float(data.get('commission', 0) or 0),
float(data.get('slippage', 0) or 0),
int(data.get('leverage', 1) or 1),
str(data.get('tradeDirection', 'long') or 'long'),
json.dumps(data.get('strategyConfig') or {}, ensure_ascii=False),
'failed',
str(e),
''
)
)
db.commit()
cur.close()
backtest_service.persist_run(
user_id=user_id,
indicator_id=int(indicator_id) if indicator_id is not None else None,
run_type='indicator',
market=str(data.get('market', '') or ''),
symbol=str(data.get('symbol', '') or ''),
timeframe=str(data.get('timeframe', '') or ''),
start_date_str=str(data.get('startDate', '') or ''),
end_date_str=str(data.get('endDate', '') or ''),
initial_capital=float(data.get('initialCapital', 0) or 0),
commission=float(data.get('commission', 0) or 0),
slippage=float(data.get('slippage', 0) or 0),
leverage=int(data.get('leverage', 1) or 1),
trade_direction=str(data.get('tradeDirection', 'long') or 'long'),
strategy_config=data.get('strategyConfig') or {},
config_snapshot={'indicatorId': int(indicator_id) if indicator_id is not None else None},
status='failed',
error_message=str(e),
result=None,
code=str(data.get('indicatorCode', '') or ''),
)
except Exception:
pass
return jsonify({
@@ -391,53 +363,22 @@ def get_backtest_history():
offset = max(0, offset)
indicator_id = request.args.get('indicatorId')
strategy_id = request.args.get('strategyId')
run_type = (request.args.get('runType') or '').strip()
symbol = (request.args.get('symbol') or '').strip()
market = (request.args.get('market') or '').strip()
timeframe = (request.args.get('timeframe') or '').strip()
where = ["user_id = ?"]
params = [user_id]
if indicator_id is not None and str(indicator_id).strip() != "":
try:
where.append("indicator_id = ?")
params.append(int(indicator_id))
except Exception:
pass
if symbol:
where.append("symbol = ?")
params.append(symbol)
if market:
where.append("market = ?")
params.append(market)
if timeframe:
where.append("timeframe = ?")
params.append(timeframe)
where_sql = " AND ".join(where)
with get_db_connection() as db:
cur = db.cursor()
cur.execute(
f"""
SELECT id, user_id, indicator_id, market, symbol, timeframe,
start_date, end_date, initial_capital, commission, slippage,
leverage, trade_direction, strategy_config, status, error_message,
created_at
FROM qd_backtest_runs
WHERE {where_sql}
ORDER BY id DESC
LIMIT ? OFFSET ?
""",
(*params, limit, offset)
)
rows = cur.fetchall() or []
cur.close()
# Parse strategy_config JSON best-effort
for r in rows:
try:
r['strategy_config'] = json.loads(r.get('strategy_config') or '{}')
except Exception:
pass
rows = backtest_service.list_runs(
user_id=user_id,
limit=limit,
offset=offset,
indicator_id=int(indicator_id) if indicator_id is not None and str(indicator_id).strip() != "" else None,
strategy_id=int(strategy_id) if strategy_id is not None and str(strategy_id).strip() != "" else None,
run_type=run_type or None,
symbol=symbol,
market=market,
timeframe=timeframe,
)
return jsonify({'code': 1, 'msg': 'OK', 'data': rows})
except Exception as e:
@@ -461,35 +402,10 @@ def get_backtest_run():
if not run_id:
return jsonify({'code': 0, 'msg': 'runId is required', 'data': None}), 400
with get_db_connection() as db:
cur = db.cursor()
cur.execute(
"""
SELECT id, user_id, indicator_id, market, symbol, timeframe,
start_date, end_date, initial_capital, commission, slippage,
leverage, trade_direction, strategy_config, status, error_message,
result_json, created_at
FROM qd_backtest_runs
WHERE id = ? AND user_id = ?
""",
(run_id, user_id),
)
row = cur.fetchone()
cur.close()
row = backtest_service.get_run(user_id=user_id, run_id=run_id)
if not row:
return jsonify({'code': 0, 'msg': 'run not found', 'data': None}), 404
try:
row['strategy_config'] = json.loads(row.get('strategy_config') or '{}')
except Exception:
pass
try:
row['result'] = json.loads(row.get('result_json') or '{}')
except Exception:
row['result'] = {}
row.pop('result_json', None)
return jsonify({'code': 1, 'msg': 'OK', 'data': row})
except Exception as e:
logger.error(f"get_backtest_run failed: {e}")
@@ -724,6 +640,7 @@ def ai_analyze_backtest_runs():
try:
data = request.get_json() or {}
user_id = g.user_id
backtest_service.ensure_storage_schema()
lang = _normalize_lang(data.get('lang'))
run_ids = data.get('runIds') or []
if not isinstance(run_ids, list) or not run_ids:
@@ -740,9 +657,9 @@ def ai_analyze_backtest_runs():
cur = db.cursor()
cur.execute(
f"""
SELECT id, user_id, indicator_id, market, symbol, timeframe,
SELECT id, user_id, indicator_id, strategy_id, strategy_name, run_type, market, symbol, timeframe,
start_date, end_date, initial_capital, commission, slippage,
leverage, trade_direction, strategy_config, status, error_message,
leverage, trade_direction, strategy_config, config_snapshot, status, error_message,
result_json, created_at
FROM qd_backtest_runs
WHERE user_id = ? AND id IN ({placeholders})
@@ -759,6 +676,10 @@ def ai_analyze_backtest_runs():
r['strategy_config'] = json.loads(r.get('strategy_config') or '{}')
except Exception:
r['strategy_config'] = {}
try:
r['config_snapshot'] = json.loads(r.get('config_snapshot') or '{}')
except Exception:
r['config_snapshot'] = {}
try:
r['result'] = json.loads(r.get('result_json') or '{}')
except Exception:
@@ -806,6 +727,9 @@ def ai_analyze_backtest_runs():
"selectedRuns": [
{
"id": r.get("id"),
"strategy_id": r.get("strategy_id"),
"strategy_name": r.get("strategy_name"),
"run_type": r.get("run_type"),
"market": r.get("market"),
"symbol": r.get("symbol"),
"timeframe": r.get("timeframe"),
@@ -814,6 +738,7 @@ def ai_analyze_backtest_runs():
"leverage": r.get("leverage"),
"trade_direction": r.get("trade_direction"),
"strategy_config": r.get("strategy_config") or {},
"config_snapshot": r.get("config_snapshot") or {},
"result": r.get("result") or {},
"status": r.get("status"),
}
@@ -833,7 +758,7 @@ def ai_analyze_backtest_runs():
{"role": "user", "content": json.dumps(user_payload, ensure_ascii=False)},
],
},
timeout=120,
timeout=30,
)
try:
resp.raise_for_status()
backend_api_python/app/routes/credentials.py
+16 -3
View File
@@ -40,7 +40,7 @@ def list_credentials():
cur = db.cursor()
cur.execute(
"""
SELECT id, user_id, name, exchange_id, api_key_hint, created_at, updated_at
SELECT id, user_id, name, exchange_id, api_key_hint, encrypted_config, created_at, updated_at
FROM qd_exchange_credentials
WHERE user_id = %s
ORDER BY id DESC
@@ -50,7 +50,20 @@ def list_credentials():
rows = cur.fetchall() or []
cur.close()
return jsonify({'code': 1, 'msg': 'success', 'data': {'items': rows}})
items = []
for row in rows:
item = dict(row or {})
item['enable_demo_trading'] = False
try:
plain = decrypt_credential_blob(item.get('encrypted_config'))
cfg = json.loads(plain) if plain else {}
item['enable_demo_trading'] = bool(cfg.get('enable_demo_trading') or cfg.get('enableDemoTrading'))
except Exception:
item['enable_demo_trading'] = False
item.pop('encrypted_config', None)
items.append(item)
return jsonify({'code': 1, 'msg': 'success', 'data': {'items': items}})
except Exception as e:
logger.error(f"list_credentials failed: {str(e)}")
logger.error(traceback.format_exc())
@@ -59,7 +72,7 @@ def list_credentials():
CRYPTO_EXCHANGES = [
'binance', 'okx', 'bitget', 'bybit', 'coinbaseexchange',
'kraken', 'kucoin', 'gate', 'bitfinex', 'deepcoin'
'kraken', 'kucoin', 'gate', 'bitfinex', 'deepcoin', 'htx'
]
backend_api_python/app/routes/quick_trade.py
+46 -8
View File
@@ -65,9 +65,17 @@ def _convert_usdt_to_base_qty(client, symbol: str, usdt_amount: float, market_ty
logger.info(f"Using limit price {limit_price} for USDT conversion")
else:
# Try to get current market price from exchange
if hasattr(client, "get_ticker"):
try:
ticker = client.get_ticker(symbol=symbol)
if isinstance(ticker, dict):
current_price = float(ticker.get("last") or ticker.get("lastPx") or ticker.get("close") or ticker.get("price") or 0)
except Exception:
current_price = 0.0
# OKX
from app.services.live_trading.okx import OkxClient
if isinstance(client, OkxClient):
if current_price <= 0 and isinstance(client, OkxClient):
try:
from app.services.live_trading.symbols import to_okx_spot_inst_id, to_okx_swap_inst_id
inst_id = to_okx_spot_inst_id(symbol) if market_type == "spot" else to_okx_swap_inst_id(symbol)
@@ -88,7 +96,7 @@ def _convert_usdt_to_base_qty(client, symbol: str, usdt_amount: float, market_ty
# Binance - try to get price from public API
from app.services.live_trading.binance import BinanceFuturesClient
from app.services.live_trading.binance_spot import BinanceSpotClient
if isinstance(client, (BinanceFuturesClient, BinanceSpotClient)):
if current_price <= 0 and isinstance(client, (BinanceFuturesClient, BinanceSpotClient)):
try:
# Binance public ticker endpoint
base_url = getattr(client, "base_url", "")
@@ -284,10 +292,10 @@ def place_order():
market_type = "spot" if leverage == 1 else "swap"
if market_type in ("futures", "future", "perp", "perpetual"):
market_type = "swap"
# Override: if leverage > 1, force swap; if leverage = 1, force spot
# Override only when user did not explicitly choose market_type.
if leverage > 1:
market_type = "swap"
elif leverage == 1:
elif leverage == 1 and not str(body.get("market_type") or "").strip():
market_type = "spot"
# ---- build exchange client ----
@@ -485,6 +493,8 @@ def _limit_order_kwargs(client, symbol, amount, price, side, market_type, client
from app.services.live_trading.binance import BinanceFuturesClient
from app.services.live_trading.binance_spot import BinanceSpotClient
from app.services.live_trading.okx import OkxClient
from app.services.live_trading.bybit import BybitClient
from app.services.live_trading.deepcoin import DeepcoinClient
if isinstance(client, (BinanceFuturesClient, BinanceSpotClient)):
return {"quantity": amount, "price": price, "client_order_id": client_order_id}
@@ -497,6 +507,8 @@ def _limit_order_kwargs(client, symbol, amount, price, side, market_type, client
pos_side = "long" if side.lower() == "buy" else "short"
kwargs["pos_side"] = pos_side
return kwargs
if isinstance(client, (BybitClient, DeepcoinClient)):
return {"qty": amount, "price": price, "client_order_id": client_order_id}
# Generic fallback
return {"size": amount, "price": price, "client_order_id": client_order_id}
@@ -592,6 +604,26 @@ def _parse_balance(raw: Any, exchange_id: str, market_type: str) -> Dict[str, An
result["available"] = float(c.get("availableToWithdraw") or c.get("walletBalance") or 0)
result["total"] = float(c.get("walletBalance") or 0)
return result
# HTX spot
if isinstance(data, dict) and isinstance(data.get("list"), list):
for item in data.get("list") or []:
if str(item.get("currency") or "").upper() == "USDT" and str(item.get("type") or "").lower() in ("trade", "available", ""):
avail = float(item.get("balance") or 0)
result["available"] = avail
total = 0.0
for item in data.get("list") or []:
if str(item.get("currency") or "").upper() == "USDT":
total += float(item.get("balance") or 0)
if total > 0 or result["available"] > 0:
result["total"] = total or result["available"]
return result
# HTX swap
if isinstance(data, list) and data and isinstance(data[0], dict):
first = data[0]
if "margin_available" in first or "margin_balance" in first or "withdraw_available" in first:
result["available"] = float(first.get("margin_available") or first.get("withdraw_available") or 0)
result["total"] = float(first.get("margin_balance") or first.get("margin_static") or 0)
return result
# Fallback: try to find any USDT-like values
if isinstance(raw, dict):
for k, v in raw.items():
@@ -679,7 +711,7 @@ def _parse_positions(raw: Any) -> list:
# SWAP: posAmt, pos
# SPOT: bal (balance), availBal (available balance)
size = float(item.get("posAmt") or item.get("pos") or item.get("size") or item.get("contracts") or
item.get("bal") or item.get("availBal") or 0)
item.get("bal") or item.get("availBal") or item.get("volume") or 0)
if abs(size) < 1e-10:
continue
@@ -693,15 +725,21 @@ def _parse_positions(raw: Any) -> list:
pos_side = str(item.get("posSide", "")).strip().lower()
if pos_side in ("long", "short"):
side = pos_side
elif item.get("direction"):
dir_side = str(item.get("direction") or "").strip().lower()
if dir_side in ("buy", "long"):
side = "long"
elif dir_side in ("sell", "short"):
side = "short"
result.append({
"symbol": item.get("symbol") or item.get("instId") or "",
"side": side,
"size": abs(size),
"entry_price": float(item.get("entryPrice") or item.get("avgCost") or item.get("avgPx") or item.get("avgPx") or 0),
"unrealized_pnl": float(item.get("unRealizedProfit") or item.get("upl") or item.get("unrealisedPnl") or item.get("pnl") or 0),
"entry_price": float(item.get("entryPrice") or item.get("avgCost") or item.get("avgPx") or item.get("cost_open") or 0),
"unrealized_pnl": float(item.get("unRealizedProfit") or item.get("upl") or item.get("unrealisedPnl") or item.get("profit_unreal") or item.get("pnl") or 0),
"leverage": float(item.get("leverage") or item.get("lever") or 1),
"mark_price": float(item.get("markPrice") or item.get("markPx") or item.get("last") or 0),
"mark_price": float(item.get("markPrice") or item.get("markPx") or item.get("last_price") or item.get("last") or 0),
})
except Exception as e:
logger.warning(f"_parse_positions error: {e}")
backend_api_python/app/routes/strategy.py
+179 -3
View File
@@ -2,12 +2,15 @@
Trading Strategy API Routes
"""
from flask import Blueprint, request, jsonify, g
from datetime import datetime
import json
import traceback
import time
from app.services.strategy import StrategyService
from app.services.strategy_compiler import StrategyCompiler
from app.services.backtest import BacktestService
from app.services.strategy_snapshot import StrategySnapshotResolver
from app import get_trading_executor
from app.utils.logger import get_logger
from app.utils.db import get_db_connection
@@ -21,6 +24,7 @@ strategy_bp = Blueprint('strategy', __name__)
# Local mode: avoid heavy initialization during module import.
# Instantiate services lazily on first use to keep startup clean.
_strategy_service = None
_backtest_service = None
def get_strategy_service() -> StrategyService:
global _strategy_service
@@ -29,6 +33,13 @@ def get_strategy_service() -> StrategyService:
return _strategy_service
def get_backtest_service() -> BacktestService:
global _backtest_service
if _backtest_service is None:
_backtest_service = BacktestService()
return _backtest_service
@strategy_bp.route('/strategies', methods=['GET'])
@login_required
def list_strategies():
@@ -63,6 +74,168 @@ def get_strategy_detail():
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
@strategy_bp.route('/strategies/backtest', methods=['POST'])
@login_required
def run_strategy_backtest():
try:
payload = request.get_json() or {}
user_id = g.user_id
strategy_id = int(payload.get('strategyId') or 0)
if not strategy_id:
return jsonify({'code': 0, 'msg': 'strategyId is required', 'data': None}), 400
start_date_str = str(payload.get('startDate') or '').strip()
end_date_str = str(payload.get('endDate') or '').strip()
if not start_date_str or not end_date_str:
return jsonify({'code': 0, 'msg': 'startDate and endDate are required', 'data': None}), 400
strategy = get_strategy_service().get_strategy(strategy_id, user_id=user_id)
if not strategy:
return jsonify({'code': 0, 'msg': 'Strategy not found', 'data': None}), 404
resolver = StrategySnapshotResolver(user_id=user_id)
snapshot = resolver.resolve(strategy, payload.get('overrideConfig') or {})
snapshot['user_id'] = user_id
start_date = datetime.strptime(start_date_str, '%Y-%m-%d')
end_date = datetime.strptime(end_date_str, '%Y-%m-%d').replace(hour=23, minute=59, second=59)
days_diff = (end_date - start_date).days
timeframe = snapshot.get('timeframe') or '1D'
if timeframe == '1m':
max_days = 30
max_range_text = '1 month'
elif timeframe == '5m':
max_days = 180
max_range_text = '6 months'
elif timeframe in ['15m', '30m']:
max_days = 365
max_range_text = '1 year'
else:
max_days = 1095
max_range_text = '3 years'
if days_diff > max_days:
return jsonify({
'code': 0,
'msg': f'Backtest range exceeds limit: timeframe {timeframe} supports up to {max_range_text} ({max_days} days), but you selected {days_diff} days',
'data': None
}), 400
svc = get_backtest_service()
result = svc.run_strategy_snapshot(snapshot, start_date=start_date, end_date=end_date)
run_id = svc.persist_run(
user_id=user_id,
indicator_id=snapshot.get('indicator_id'),
strategy_id=snapshot.get('strategy_id'),
strategy_name=snapshot.get('strategy_name') or '',
run_type=snapshot.get('run_type') or 'strategy_indicator',
market=snapshot.get('market') or '',
symbol=snapshot.get('symbol') or '',
timeframe=snapshot.get('timeframe') or '',
start_date_str=start_date_str,
end_date_str=end_date_str,
initial_capital=float(snapshot.get('initial_capital') or 0),
commission=float(snapshot.get('commission') or 0),
slippage=float(snapshot.get('slippage') or 0),
leverage=int(snapshot.get('leverage') or 1),
trade_direction=str(snapshot.get('trade_direction') or 'long'),
strategy_config=snapshot.get('strategy_config') or {},
config_snapshot=snapshot.get('config_snapshot') or {},
status='success',
error_message='',
result=result,
code=snapshot.get('code') or '',
)
return jsonify({'code': 1, 'msg': 'success', 'data': {'runId': run_id, 'result': result}})
except ValueError as e:
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 400
except Exception as e:
logger.error(f"run_strategy_backtest failed: {str(e)}")
logger.error(traceback.format_exc())
try:
payload = payload if isinstance(payload, dict) else {}
strategy_id = int(payload.get('strategyId') or 0)
strategy = get_strategy_service().get_strategy(strategy_id, user_id=g.user_id) if strategy_id else None
if strategy:
resolver = StrategySnapshotResolver(user_id=g.user_id)
snapshot = resolver.resolve(strategy, payload.get('overrideConfig') or {})
snapshot['user_id'] = g.user_id
get_backtest_service().persist_run(
user_id=g.user_id,
indicator_id=snapshot.get('indicator_id'),
strategy_id=snapshot.get('strategy_id'),
strategy_name=snapshot.get('strategy_name') or '',
run_type=snapshot.get('run_type') or 'strategy_indicator',
market=snapshot.get('market') or '',
symbol=snapshot.get('symbol') or '',
timeframe=snapshot.get('timeframe') or '',
start_date_str=str(payload.get('startDate') or ''),
end_date_str=str(payload.get('endDate') or ''),
initial_capital=float(snapshot.get('initial_capital') or 0),
commission=float(snapshot.get('commission') or 0),
slippage=float(snapshot.get('slippage') or 0),
leverage=int(snapshot.get('leverage') or 1),
trade_direction=str(snapshot.get('trade_direction') or 'long'),
strategy_config=snapshot.get('strategy_config') or {},
config_snapshot=snapshot.get('config_snapshot') or {},
status='failed',
error_message=str(e),
result=None,
code=snapshot.get('code') or '',
)
except Exception:
pass
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
@strategy_bp.route('/strategies/backtest/history', methods=['GET'])
@login_required
def get_strategy_backtest_history():
try:
user_id = g.user_id
strategy_id = int(request.args.get('strategyId') or request.args.get('id') or 0)
if not strategy_id:
return jsonify({'code': 0, 'msg': 'strategyId is required', 'data': None}), 400
limit = max(1, min(int(request.args.get('limit') or 50), 200))
offset = max(0, int(request.args.get('offset') or 0))
symbol = (request.args.get('symbol') or '').strip()
market = (request.args.get('market') or '').strip()
timeframe = (request.args.get('timeframe') or '').strip()
rows = get_backtest_service().list_runs(
user_id=user_id,
strategy_id=strategy_id,
limit=limit,
offset=offset,
symbol=symbol,
market=market,
timeframe=timeframe,
)
rows = [r for r in rows if str(r.get('run_type') or '').startswith('strategy_')]
return jsonify({'code': 1, 'msg': 'success', 'data': rows})
except Exception as e:
logger.error(f"get_strategy_backtest_history failed: {str(e)}")
logger.error(traceback.format_exc())
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
@strategy_bp.route('/strategies/backtest/get', methods=['GET'])
@login_required
def get_strategy_backtest_run():
try:
user_id = g.user_id
run_id = int(request.args.get('runId') or 0)
if not run_id:
return jsonify({'code': 0, 'msg': 'runId is required', 'data': None}), 400
row = get_backtest_service().get_run(user_id=user_id, run_id=run_id)
if not row or not str(row.get('run_type') or '').startswith('strategy_'):
return jsonify({'code': 0, 'msg': 'run not found', 'data': None}), 404
return jsonify({'code': 1, 'msg': 'success', 'data': row})
except Exception as e:
logger.error(f"get_strategy_backtest_run failed: {str(e)}")
logger.error(traceback.format_exc())
return jsonify({'code': 0, 'msg': str(e), 'data': None}), 500
@strategy_bp.route('/strategies/create', methods=['POST'])
@login_required
def create_strategy():
@@ -695,7 +868,7 @@ def test_connection():
return jsonify({'code': 0, 'msg': 'Please provide API key and secret key', 'data': None})
# Pass the resolved config (with actual keys) to the service
result = get_strategy_service().test_exchange_connection(resolved)
result = get_strategy_service().test_exchange_connection(resolved, user_id=user_id)
if result['success']:
return jsonify({'code': 1, 'msg': result.get('message') or 'Connection successful', 'data': result.get('data')})
@@ -1079,9 +1252,12 @@ def ai_generate_strategy():
Generate Python strategy code that follows this framework:
- def on_init(ctx): Initialize strategy parameters using ctx.param(name, default)
- def on_bar(ctx, bar): Core logic called on each K-line bar
- bar has: open, high, low, close, volume, timestamp
- bar supports both bar.close and bar['close'] access, and has: open, high, low, close, volume, timestamp
- ctx.buy(price, amount), ctx.sell(price, amount), ctx.close_position()
- ctx.position (current position), ctx.balance, ctx.equity
- ctx.position supports both numeric checks and dict-style fields:
- if not ctx.position / if ctx.position > 0 / if ctx.position < 0
- ctx.position['side'], ctx.position['size'], ctx.position['entry_price']
- ctx.balance, ctx.equity
- ctx.bars(n) to get last N bars, ctx.log(message) to log
- def on_order_filled(ctx, order): Optional callback when order fills
- def on_stop(ctx): Optional cleanup when strategy stops
backend_api_python/app/routes/user.py
+44 -1
View File
@@ -873,16 +873,48 @@ def get_system_strategies():
page: int (default 1)
page_size: int (default 20, max 100)
status: str (optional, filter by status: running/stopped/all)
execution_mode: str (optional, live/signal — omit or all for any)
search: str (optional, search by strategy name/symbol/username)
sort_by: str (optional, whitelist; default status+updated_at)
sort_order: str (optional, asc or desc; default desc when sort_by set)
"""
try:
page = request.args.get('page', 1, type=int)
page_size = request.args.get('page_size', 20, type=int)
status_filter = request.args.get('status', '', type=str).strip().lower()
execution_filter = request.args.get('execution_mode', '', type=str).strip().lower()
search = request.args.get('search', '', type=str).strip()
sort_by = request.args.get('sort_by', '', type=str).strip().lower()
sort_order = request.args.get('sort_order', 'desc', type=str).strip().lower()
if sort_order not in ('asc', 'desc'):
sort_order = 'desc'
page_size = min(100, max(1, page_size))
offset = (page - 1) * page_size
sort_sql_map = {
'id': 's.id',
'updated_at': 's.updated_at',
'created_at': 's.created_at',
'initial_capital': 's.initial_capital',
'strategy_name': 's.strategy_name',
'symbol': 's.symbol',
'status': 's.status',
'execution_mode': 's.execution_mode',
'leverage': 's.leverage',
}
sort_expr_map = {
'total_pnl': (
"(COALESCE((SELECT SUM(unrealized_pnl) FROM qd_strategy_positions p WHERE p.strategy_id = s.id), 0)"
" + COALESCE((SELECT SUM(profit) FROM qd_strategy_trades t WHERE t.strategy_id = s.id), 0))"
),
'trade_count': '(SELECT COUNT(*) FROM qd_strategy_trades t WHERE t.strategy_id = s.id)',
'position_count': '(SELECT COUNT(*) FROM qd_strategy_positions p WHERE p.strategy_id = s.id)',
'total_equity': (
'COALESCE((SELECT SUM(equity) FROM qd_strategy_positions p WHERE p.strategy_id = s.id), 0)'
),
}
direction = 'ASC' if sort_order == 'asc' else 'DESC'
with get_db_connection() as db:
cur = db.cursor()
@@ -894,6 +926,10 @@ def get_system_strategies():
conditions.append("s.status = ?")
params.append(status_filter)
if execution_filter in ('live', 'signal'):
conditions.append("s.execution_mode = ?")
params.append(execution_filter)
if search:
conditions.append(
"(s.strategy_name ILIKE ? OR s.symbol ILIKE ? OR u.username ILIKE ? OR u.nickname ILIKE ?)"
@@ -905,6 +941,13 @@ def get_system_strategies():
if conditions:
where_clause = "WHERE " + " AND ".join(conditions)
if sort_by in sort_sql_map:
order_clause = f"ORDER BY {sort_sql_map[sort_by]} {direction}, s.id DESC"
elif sort_by in sort_expr_map:
order_clause = f"ORDER BY {sort_expr_map[sort_by]} {direction}, s.id DESC"
else:
order_clause = "ORDER BY s.status DESC, s.updated_at DESC, s.id DESC"
# Get total count
count_sql = f"""
SELECT COUNT(*) as cnt
@@ -941,7 +984,7 @@ def get_system_strategies():
FROM qd_strategies_trading s
LEFT JOIN qd_users u ON u.id = s.user_id
{where_clause}
ORDER BY s.status DESC, s.updated_at DESC
{order_clause}
LIMIT ? OFFSET ?
"""
cur.execute(query_sql, tuple(params) + (page_size, offset))