stable/freqtrade/rpc/api_server/api_v1.py

240 lines
8.5 KiB
Python
Raw Normal View History

2020-12-26 19:05:27 +00:00
from copy import deepcopy
from pathlib import Path
2020-12-27 08:02:35 +00:00
from typing import Dict, List, Optional, Union
2020-12-26 15:43:15 +00:00
from fastapi import APIRouter, Depends
2020-12-26 19:05:27 +00:00
from fastapi.exceptions import HTTPException
2020-12-25 12:11:01 +00:00
from freqtrade import __version__
2020-12-27 08:02:35 +00:00
from freqtrade.constants import USERPATH_STRATEGIES
2020-12-26 19:05:27 +00:00
from freqtrade.data.history import get_datahandler
from freqtrade.exceptions import OperationalException
2020-12-26 14:54:22 +00:00
from freqtrade.rpc import RPC
2021-01-02 08:07:31 +00:00
from freqtrade.rpc.api_server.api_schemas import (AvailablePairs, Balances, BlacklistPayload,
BlacklistResponse, Count, Daily, DeleteTrade,
ForceBuyPayload, ForceSellPayload, Locks, Logs,
OpenTradeSchema, PairHistory, PerformanceEntry,
Ping, PlotConfig, Profit, ResultMsg, Stats,
StatusMsg, StrategyListResponse, StrategyResponse,
TradeResponse, TradeSchema, Version,
WhitelistResponse)
2021-01-02 12:12:49 +00:00
from freqtrade.rpc.api_server.deps import get_config, get_rpc, get_rpc_optional
from freqtrade.rpc.rpc import RPCException
2020-12-25 12:11:01 +00:00
# Public API, requires no auth.
router_public = APIRouter()
2020-12-25 12:11:01 +00:00
# Private API, protected by authentication
router = APIRouter()
@router_public.get('/ping', response_model=Ping)
def ping():
2020-12-26 16:48:19 +00:00
"""simple ping"""
return {"status": "pong"}
2020-12-25 19:07:12 +00:00
@router.get('/version', response_model=Version, tags=['info'])
2020-12-25 14:57:05 +00:00
def version():
2020-12-26 16:48:19 +00:00
""" Bot Version info"""
2020-12-25 14:57:05 +00:00
return {"version": __version__}
2020-12-25 19:07:12 +00:00
@router.get('/balance', response_model=Balances, tags=['info'])
def balance(rpc: RPC = Depends(get_rpc), config=Depends(get_config)):
2020-12-26 16:48:19 +00:00
"""Account Balances"""
return rpc._rpc_balance(config['stake_currency'], config.get('fiat_display_currency', ''),)
2020-12-25 12:11:01 +00:00
2020-12-26 14:54:22 +00:00
@router.get('/count', response_model=Count, tags=['info'])
def count(rpc: RPC = Depends(get_rpc)):
return rpc._rpc_count()
2020-12-26 15:43:15 +00:00
@router.get('/performance', response_model=List[PerformanceEntry], tags=['info'])
def performance(rpc: RPC = Depends(get_rpc)):
return rpc._rpc_performance()
@router.get('/profit', response_model=Profit, tags=['info'])
def profit(rpc: RPC = Depends(get_rpc), config=Depends(get_config)):
return rpc._rpc_trade_statistics(config['stake_currency'],
config.get('fiat_display_currency')
)
@router.get('/stats', response_model=Stats, tags=['info'])
def stats(rpc: RPC = Depends(get_rpc)):
return rpc._rpc_stats()
2020-12-26 16:33:27 +00:00
@router.get('/daily', response_model=Daily, tags=['info'])
def daily(timescale: int = 7, rpc: RPC = Depends(get_rpc), config=Depends(get_config)):
return rpc._rpc_daily_profit(timescale, config['stake_currency'],
config.get('fiat_display_currency', ''))
2021-01-01 18:38:28 +00:00
@router.get('/status', response_model=List[OpenTradeSchema], tags=['info'])
def status(rpc: RPC = Depends(get_rpc)):
try:
return rpc._rpc_trade_status()
except RPCException:
return []
2021-01-01 18:36:03 +00:00
@router.get('/trades', response_model=TradeResponse, tags=['info', 'trading'])
2020-12-31 10:01:50 +00:00
def trades(limit: int = 0, rpc: RPC = Depends(get_rpc)):
2020-12-26 16:33:27 +00:00
return rpc._rpc_trade_history(limit)
@router.delete('/trades/{tradeid}', response_model=DeleteTrade, tags=['info', 'trading'])
def trades_delete(tradeid: int, rpc: RPC = Depends(get_rpc)):
return rpc._rpc_delete(tradeid)
# TODO: Missing response model
@router.get('/edge', tags=['info'])
def edge(rpc: RPC = Depends(get_rpc)):
return rpc._rpc_edge()
# TODO: Missing response model
2020-12-26 14:54:22 +00:00
@router.get('/show_config', tags=['info'])
2021-01-02 12:12:49 +00:00
def show_config(rpc: Optional[RPC] = Depends(get_rpc_optional), config=Depends(get_config)):
state = ''
if rpc:
state = rpc._freqtrade.state
return RPC._rpc_show_config(config, state)
2020-12-26 14:54:22 +00:00
2020-12-26 16:33:27 +00:00
2021-01-01 18:36:03 +00:00
@router.post('/forcebuy', response_model=Union[TradeSchema, StatusMsg], tags=['trading'])
def forcebuy(payload: ForceBuyPayload, rpc: RPC = Depends(get_rpc)):
trade = rpc._rpc_forcebuy(payload.pair, payload.price)
if trade:
return trade.to_json()
else:
return {"status": f"Error buying pair {payload.pair}."}
2020-12-26 16:33:27 +00:00
@router.post('/forcesell', response_model=ResultMsg, tags=['trading'])
def forcesell(payload: ForceSellPayload, rpc: RPC = Depends(get_rpc)):
return rpc._rpc_forcesell(payload.tradeid)
2020-12-26 14:54:22 +00:00
2020-12-26 16:33:27 +00:00
@router.get('/blacklist', response_model=BlacklistResponse, tags=['info', 'pairlist'])
def blacklist(rpc: RPC = Depends(get_rpc)):
return rpc._rpc_blacklist()
@router.post('/blacklist', response_model=BlacklistResponse, tags=['info', 'pairlist'])
def blacklist_post(payload: BlacklistPayload, rpc: RPC = Depends(get_rpc)):
return rpc._rpc_blacklist(payload.blacklist)
@router.get('/whitelist', response_model=WhitelistResponse, tags=['info', 'pairlist'])
def whitelist(rpc: RPC = Depends(get_rpc)):
return rpc._rpc_whitelist()
@router.get('/locks', response_model=Locks, tags=['info'])
def locks(rpc: RPC = Depends(get_rpc)):
return rpc._rpc_locks()
@router.get('/logs', response_model=Logs, tags=['info'])
def logs(limit: Optional[int] = None, rpc: RPC = Depends(get_rpc)):
return rpc._rpc_get_logs(limit)
2020-12-25 19:07:12 +00:00
@router.post('/start', response_model=StatusMsg, tags=['botcontrol'])
def start(rpc: RPC = Depends(get_rpc)):
return rpc._rpc_start()
@router.post('/stop', response_model=StatusMsg, tags=['botcontrol'])
def stop(rpc: RPC = Depends(get_rpc)):
return rpc._rpc_stop()
2020-12-26 14:54:22 +00:00
@router.post('/stopbuy', response_model=StatusMsg, tags=['botcontrol'])
def stop_buy(rpc: RPC = Depends(get_rpc)):
return rpc._rpc_stopbuy()
@router.post('/reload_config', response_model=StatusMsg, tags=['botcontrol'])
def reload_config(rpc: RPC = Depends(get_rpc)):
return rpc._rpc_reload_config()
2020-12-26 19:05:27 +00:00
2020-12-27 08:02:35 +00:00
@router.get('/pair_candles', response_model=PairHistory, tags=['candle data'])
2020-12-26 19:05:27 +00:00
def pair_candles(pair: str, timeframe: str, limit: Optional[int], rpc=Depends(get_rpc)):
return rpc._rpc_analysed_dataframe(pair, timeframe, limit)
2020-12-27 08:02:35 +00:00
@router.get('/pair_history', response_model=PairHistory, tags=['candle data'])
2020-12-26 19:05:27 +00:00
def pair_history(pair: str, timeframe: str, timerange: str, strategy: str,
config=Depends(get_config)):
config = deepcopy(config)
config.update({
'strategy': strategy,
})
return RPC._rpc_analysed_history_full(config, pair, timeframe, timerange)
2020-12-31 10:01:50 +00:00
@router.get('/plot_config', response_model=Union[PlotConfig, Dict], tags=['candle data'])
def plot_config(rpc: RPC = Depends(get_rpc)):
2020-12-26 19:05:27 +00:00
return rpc._rpc_plot_config()
@router.get('/strategies', response_model=StrategyListResponse, tags=['strategy'])
def list_strategies(config=Depends(get_config)):
directory = Path(config.get(
'strategy_path', config['user_data_dir'] / USERPATH_STRATEGIES))
from freqtrade.resolvers.strategy_resolver import StrategyResolver
strategies = StrategyResolver.search_all_objects(directory, False)
strategies = sorted(strategies, key=lambda x: x['name'])
return {'strategies': [x['name'] for x in strategies]}
@router.get('/strategy/{strategy}', response_model=StrategyResponse, tags=['strategy'])
def get_strategy(strategy: str, rpc: RPC = Depends(get_rpc), config=Depends(get_config)):
config = deepcopy(config)
from freqtrade.resolvers.strategy_resolver import StrategyResolver
try:
strategy_obj = StrategyResolver._load_strategy(strategy, config,
extra_dir=config.get('strategy_path'))
except OperationalException:
raise HTTPException(status_code=404, detail='Strategy not found')
return {
'strategy': strategy_obj.get_strategy_name(),
'code': strategy_obj.__source__,
}
@router.get('/available_pairs', response_model=AvailablePairs, tags=['candle data'])
def list_available_pairs(timeframe: Optional[str] = None, stake_currency: Optional[str] = None,
config=Depends(get_config)):
dh = get_datahandler(config['datadir'], config.get('dataformat_ohlcv', None))
pair_interval = dh.ohlcv_get_available_data(config['datadir'])
if timeframe:
pair_interval = [pair for pair in pair_interval if pair[1] == timeframe]
if stake_currency:
pair_interval = [pair for pair in pair_interval if pair[0].endswith(stake_currency)]
pair_interval = sorted(pair_interval, key=lambda x: x[0])
pairs = list({x[0] for x in pair_interval})
result = {
'length': len(pairs),
'pairs': pairs,
'pair_interval': pair_interval,
}
return result