Timeframe
5m
Direction
Long Only
Stoploss
-5.0%
Trailing Stop
No
ROI
N/A
Interface Version
N/A
Startup Candles
N/A
Indicators
0
freqtrade/freqtrade-strategies
author@: lenik
# backend/routes/custom_strategies.py
"""
Custom Strategy API Routes
Handles strategy upload, backtesting, hyperopt, and deployment
"""
from fastapi import APIRouter, UploadFile, File, HTTPException, Header
from fastapi.responses import JSONResponse
from typing import Optional
from datetime import datetime
import logging
from ..integrations.freqtrade_integration import FreqtradeBacktester
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/yuki/strategies/custom", tags=["custom_strategies"])
# Initialize backtester (in production: inject via dependency)
backtester = FreqtradeBacktester(base_config={})
@router.post("/upload")
async def upload_strategy(
file: UploadFile = File(...),
description: Optional[str] = None
):
"""
Upload custom strategy file (Freqtrade format)
Supports:
- Freqtrade IStrategy subclass
- Custom Python agent
Returns:
strategy_id for use in other endpoints
"""
try:
content = await file.read()
strategy_code = content.decode('utf-8')
# 1. Validate syntax
is_valid, error = backtester.validate_strategy_syntax(strategy_code)
if not is_valid:
raise HTTPException(status_code=400, detail=error)
# 2. Check for Freqtrade methods
is_freqtrade, missing = backtester.validate_freqtrade_interface(strategy_code)
# 3. Store in database (TODO)
strategy_id = f"strat_{datetime.now().timestamp()}".replace(".", "_")
logger.info(f"Strategy uploaded: {strategy_id} from {file.filename}")
return {
"status": "success",
"strategy_id": strategy_id,
"filename": file.filename,
"is_freqtrade_format": is_freqtrade,
"description": description,
"uploaded_at": datetime.now().isoformat()
}
except UnicodeDecodeError:
raise HTTPException(status_code=400, detail="File must be valid UTF-8")
except Exception as e:
logger.error(f"Upload error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.post("/backtest/{strategy_id}")
async def backtest_strategy(
strategy_id: str,
timerange: str = "20230101-20240101",
stake_amount: float = 100,
pair: str = "SOL/USDC",
timeframe: str = "5m"
):
"""
Backtest uploaded strategy on historical data
Args:
strategy_id: Strategy identifier from upload
timerange: Format "YYYYMMDD-YYYYMMDD"
stake_amount: Initial capital per trade in USDC
pair: Trading pair
timeframe: Candle timeframe
Returns:
Backtest results with metrics
"""
try:
# TODO: Retrieve strategy code from database
strategy_code = """
# Placeholder strategy
from freqtrade.strategy import IStrategy
import pandas as pd
class PlaceholderStrategy(IStrategy):
stoploss = -0.05
timeframe = "5m"
def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
return dataframe
def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
dataframe.loc[dataframe.index > 0, 'enter_long'] = 1
return dataframe
def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
dataframe.loc[dataframe.index > 5, 'exit_long'] = 1
return dataframe
"""
result = backtester.backtest_strategy(
strategy_code=strategy_code,
strategy_id=strategy_id,
timerange=timerange,
stake_amount=stake_amount,
pair=pair,
timeframe=timeframe
)
return result
except Exception as e:
logger.error(f"Backtest error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.post("/hyperopt/{strategy_id}")
async def hyperopt_strategy(
strategy_id: str,
n_epochs: int = 100,
timerange: str = "20230101-20240101",
loss_function: str = "SharpeHyperOptLoss"
):
"""
Run hyperparameter optimization on strategy
Finds best parameters to maximize specified loss function
Returns:
Best parameters found + top 3 results
"""
try:
# TODO: Retrieve strategy code from database
strategy_code = "# Strategy code here"
result = backtester.hyperopt_strategy(
strategy_code=strategy_code,
strategy_id=strategy_id,
n_epochs=n_epochs,
timerange=timerange,
loss_function=loss_function
)
return result
except Exception as e:
logger.error(f"Hyperopt error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.post("/deploy/{strategy_id}")
async def deploy_strategy(
strategy_id: str,
mode: str = "paper", # paper | live
initial_balance: float = 1000,
max_open_trades: int = 3
):
"""
Deploy strategy to live or paper trading
Returns:
Deployment status and connection details
"""
try:
# TODO: Actual deployment logic
return {
"status": "success",
"strategy_id": strategy_id,
"mode": mode,
"initial_balance": initial_balance,
"max_open_trades": max_open_trades,
"deployed_at": datetime.now().isoformat(),
"websocket_url": "wss://yuki.api/ws/strategy/signals"
}
except Exception as e:
logger.error(f"Deploy error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/{strategy_id}")
async def get_strategy(strategy_id: str):
"""Get strategy metadata and status"""
try:
result = backtester.get_strategy_info(strategy_id)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("")
async def list_strategies():
"""List all uploaded strategies"""
try:
result = backtester.list_strategies()
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))