9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.68

/freqtrade/rpc/api_server/api_v1.py

import logging
from copy import deepcopy
from typing import List, Optional

from fastapi import APIRouter, Depends, Query
from fastapi.exceptions import HTTPException

from freqtrade import __version__
from freqtrade.data.history import get_datahandler
from freqtrade.enums import CandleType, TradingMode
from freqtrade.exceptions import OperationalException
from freqtrade.rpc import RPC
from freqtrade.rpc.api_server.api_schemas import (AvailablePairs, Balances, BlacklistPayload,
                                                  BlacklistResponse, Count, DailyWeeklyMonthly,
                                                  DeleteLockRequest, DeleteTrade, Entry,
                                                  ExchangeListResponse, Exit, ForceEnterPayload,
                                                  ForceEnterResponse, ForceExitPayload,
                                                  FreqAIModelListResponse, Health, Locks,
                                                  LocksPayload, Logs, MixTag, OpenTradeSchema,
                                                  PairHistory, PerformanceEntry, Ping, PlotConfig,
                                                  Profit, ResultMsg, ShowConfig, Stats, StatusMsg,
                                                  StrategyListResponse, StrategyResponse, SysInfo,
                                                  Version, WhitelistResponse)
from freqtrade.rpc.api_server.deps import get_config, get_exchange, get_rpc, get_rpc_optional
from freqtrade.rpc.rpc import RPCException


logger = logging.getLogger(__name__)

# API version
# Pre-1.1, no version was provided
# Version increments should happen in "small" steps (1.1, 1.12, ...) unless big changes happen.
# 1.11: forcebuy and forcesell accept ordertype
# 1.12: add blacklist delete endpoint
# 1.13: forcebuy supports stake_amount
# versions 2.xx -> futures/short branch
# 2.14: Add entry/exit orders to trade response
# 2.15: Add backtest history endpoints
# 2.16: Additional daily metrics
# 2.17: Forceentry - leverage, partial force_exit
# 2.20: Add websocket endpoints
# 2.21: Add new_candle messagetype
# 2.22: Add FreqAI to backtesting
# 2.23: Allow plot config request in webserver mode
# 2.24: Add cancel_open_order endpoint
# 2.25: Add several profit values to /status endpoint
# 2.26: increase /balance output
# 2.27: Add /trades/<id>/reload endpoint
# 2.28: Switch reload endpoint to Post
# 2.29: Add /exchanges endpoint
# 2.30: new /pairlists endpoint
# 2.31: new /backtest/history/ delete endpoint
# 2.32: new /backtest/history/ patch endpoint
# 2.33: Additional weekly/monthly metrics
# 2.34: new entries/exits/mix_tags endpoints
API_VERSION = 2.34

# Public API, requires no auth.
router_public = APIRouter()
# Private API, protected by authentication
router = APIRouter()


@router_public.get('/ping', response_model=Ping)
def ping():
    """simple ping"""
    return {"status": "pong"}


@router.get('/version', response_model=Version, tags=['info'])
def version():
    """ Bot Version info"""
    return {"version": __version__}


@router.get('/balance', response_model=Balances, tags=['info'])
def balance(rpc: RPC = Depends(get_rpc), config=Depends(get_config)):
    """Account Balances"""
    return rpc._rpc_balance(config['stake_currency'], config.get('fiat_display_currency', ''),)


@router.get('/count', response_model=Count, tags=['info'])
def count(rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_count()


@router.get('/entries', response_model=List[Entry], tags=['info'])
def entries(pair: Optional[str] = None, rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_enter_tag_performance(pair)


@router.get('/exits', response_model=List[Exit], tags=['info'])
def exits(pair: Optional[str] = None, rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_exit_reason_performance(pair)


@router.get('/mix_tags', response_model=List[MixTag], tags=['info'])
def mix_tags(pair: Optional[str] = None, rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_mix_tag_performance(pair)


@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()


@router.get('/daily', response_model=DailyWeeklyMonthly, tags=['info'])
def daily(timescale: int = 7, rpc: RPC = Depends(get_rpc), config=Depends(get_config)):
    return rpc._rpc_timeunit_profit(timescale, config['stake_currency'],
                                    config.get('fiat_display_currency', ''))


@router.get('/weekly', response_model=DailyWeeklyMonthly, tags=['info'])
def weekly(timescale: int = 4, rpc: RPC = Depends(get_rpc), config=Depends(get_config)):
    return rpc._rpc_timeunit_profit(timescale, config['stake_currency'],
                                    config.get('fiat_display_currency', ''), 'weeks')


@router.get('/monthly', response_model=DailyWeeklyMonthly, tags=['info'])
def monthly(timescale: int = 3, rpc: RPC = Depends(get_rpc), config=Depends(get_config)):
    return rpc._rpc_timeunit_profit(timescale, config['stake_currency'],
                                    config.get('fiat_display_currency', ''), 'months')


@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 []


# Using the responsemodel here will cause a ~100% increase in response time (from 1s to 2s)
# on big databases. Correct response model: response_model=TradeResponse,
@router.get('/trades', tags=['info', 'trading'])
def trades(limit: int = 500, offset: int = 0, rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_trade_history(limit, offset=offset, order_by_id=True)


@router.get('/trade/{tradeid}', response_model=OpenTradeSchema, tags=['info', 'trading'])
def trade(tradeid: int = 0, rpc: RPC = Depends(get_rpc)):
    try:
        return rpc._rpc_trade_status([tradeid])[0]
    except (RPCException, KeyError):
        raise HTTPException(status_code=404, detail='Trade not found.')


@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)


@router.delete('/trades/{tradeid}/open-order', response_model=OpenTradeSchema,  tags=['trading'])
def trade_cancel_open_order(tradeid: int, rpc: RPC = Depends(get_rpc)):
    rpc._rpc_cancel_open_order(tradeid)
    return rpc._rpc_trade_status([tradeid])[0]


@router.post('/trades/{tradeid}/reload', response_model=OpenTradeSchema,  tags=['trading'])
def trade_reload(tradeid: int, rpc: RPC = Depends(get_rpc)):
    rpc._rpc_reload_trade_from_exchange(tradeid)
    return rpc._rpc_trade_status([tradeid])[0]


# TODO: Missing response model
@router.get('/edge', tags=['info'])
def edge(rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_edge()


@router.get('/show_config', response_model=ShowConfig, tags=['info'])
def show_config(rpc: Optional[RPC] = Depends(get_rpc_optional), config=Depends(get_config)):
    state = ''
    strategy_version = None
    if rpc:
        state = rpc._freqtrade.state
        strategy_version = rpc._freqtrade.strategy.version()
    resp = RPC._rpc_show_config(config, state, strategy_version)
    resp['api_version'] = API_VERSION
    return resp


# /forcebuy is deprecated with short addition. use /forceentry instead
@router.post('/forceenter', response_model=ForceEnterResponse, tags=['trading'])
@router.post('/forcebuy', response_model=ForceEnterResponse, tags=['trading'])
def force_entry(payload: ForceEnterPayload, rpc: RPC = Depends(get_rpc)):
    ordertype = payload.ordertype.value if payload.ordertype else None

    trade = rpc._rpc_force_entry(payload.pair, payload.price, order_side=payload.side,
                                 order_type=ordertype, stake_amount=payload.stakeamount,
                                 enter_tag=payload.entry_tag or 'force_entry',
                                 leverage=payload.leverage)

    if trade:
        return ForceEnterResponse.model_validate(trade.to_json())
    else:
        return ForceEnterResponse.model_validate(
            {"status": f"Error entering {payload.side} trade for pair {payload.pair}."})


# /forcesell is deprecated with short addition. use /forceexit instead
@router.post('/forceexit', response_model=ResultMsg, tags=['trading'])
@router.post('/forcesell', response_model=ResultMsg, tags=['trading'])
def forceexit(payload: ForceExitPayload, rpc: RPC = Depends(get_rpc)):
    ordertype = payload.ordertype.value if payload.ordertype else None
    return rpc._rpc_force_exit(str(payload.tradeid), ordertype, amount=payload.amount)


@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.delete('/blacklist', response_model=BlacklistResponse, tags=['info', 'pairlist'])
def blacklist_delete(pairs_to_delete: List[str] = Query([]), rpc: RPC = Depends(get_rpc)):
    """Provide a list of pairs to delete from the blacklist"""

    return rpc._rpc_blacklist_delete(pairs_to_delete)


@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', 'locks'])
def locks(rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_locks()


@router.delete('/locks/{lockid}', response_model=Locks, tags=['info', 'locks'])
def delete_lock(lockid: int, rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_delete_lock(lockid=lockid)


@router.post('/locks/delete', response_model=Locks, tags=['info', 'locks'])
def delete_lock_pair(payload: DeleteLockRequest, rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_delete_lock(lockid=payload.lockid, pair=payload.pair)


@router.post('/locks', response_model=Locks, tags=['info', 'locks'])
def add_locks(payload: List[LocksPayload], rpc: RPC = Depends(get_rpc)):
    for lock in payload:
        rpc._rpc_add_lock(lock.pair, lock.until, lock.reason, lock.side)
    return rpc._rpc_locks()


@router.get('/logs', response_model=Logs, tags=['info'])
def logs(limit: Optional[int] = None):
    return RPC._rpc_get_logs(limit)


@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()


@router.post('/stopentry', response_model=StatusMsg, tags=['botcontrol'])
@router.post('/stopbuy', response_model=StatusMsg, tags=['botcontrol'])
def stop_buy(rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_stopentry()


@router.post('/reload_config', response_model=StatusMsg, tags=['botcontrol'])
def reload_config(rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_reload_config()


@router.get('/pair_candles', response_model=PairHistory, tags=['candle data'])
def pair_candles(
        pair: str, timeframe: str, limit: Optional[int] = None, rpc: RPC = Depends(get_rpc)):
    return rpc._rpc_analysed_dataframe(pair, timeframe, limit)


@router.get('/pair_history', response_model=PairHistory, tags=['candle data'])
def pair_history(pair: str, timeframe: str, timerange: str, strategy: str,
                 freqaimodel: Optional[str] = None,
                 config=Depends(get_config), exchange=Depends(get_exchange)):
    # The initial call to this endpoint can be slow, as it may need to initialize
    # the exchange class.
    config = deepcopy(config)
    config.update({
        'strategy': strategy,
        'timerange': timerange,
        'freqaimodel': freqaimodel if freqaimodel else config.get('freqaimodel'),
    })
    try:
        return RPC._rpc_analysed_history_full(config, pair, timeframe, exchange)
    except Exception as e:
        raise HTTPException(status_code=502, detail=str(e))


@router.get('/plot_config', response_model=PlotConfig, tags=['candle data'])
def plot_config(strategy: Optional[str] = None, config=Depends(get_config),
                rpc: Optional[RPC] = Depends(get_rpc_optional)):
    if not strategy:
        if not rpc:
            raise RPCException("Strategy is mandatory in webserver mode.")
        return PlotConfig.model_validate(rpc._rpc_plot_config())
    else:
        config1 = deepcopy(config)
        config1.update({
            'strategy': strategy
        })
    try:
        return PlotConfig.model_validate(RPC._rpc_plot_config_with_strategy(config1))
    except Exception as e:
        raise HTTPException(status_code=502, detail=str(e))


@router.get('/strategies', response_model=StrategyListResponse, tags=['strategy'])
def list_strategies(config=Depends(get_config)):
    from freqtrade.resolvers.strategy_resolver import StrategyResolver
    strategies = StrategyResolver.search_all_objects(
        config, False, config.get('recursive_strategy_search', 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, config=Depends(get_config)):
    if ":" in strategy:
        raise HTTPException(status_code=500, detail="base64 encoded strategies are not allowed.")

    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')
    except Exception as e:
        raise HTTPException(status_code=502, detail=str(e))
    return {
        'strategy': strategy_obj.get_strategy_name(),
        'code': strategy_obj.__source__,
        'timeframe': getattr(strategy_obj, 'timeframe', None),
    }


@router.get('/exchanges', response_model=ExchangeListResponse, tags=[])
def list_exchanges(config=Depends(get_config)):
    from freqtrade.exchange import list_available_exchanges
    exchanges = list_available_exchanges(config)
    return {
        'exchanges': exchanges,
    }


@router.get('/freqaimodels', response_model=FreqAIModelListResponse, tags=['freqai'])
def list_freqaimodels(config=Depends(get_config)):
    from freqtrade.resolvers.freqaimodel_resolver import FreqaiModelResolver
    models = FreqaiModelResolver.search_all_objects(
        config, False)
    models = sorted(models, key=lambda x: x['name'])

    return {'freqaimodels': [x['name'] for x in models]}


@router.get('/available_pairs', response_model=AvailablePairs, tags=['candle data'])
def list_available_pairs(timeframe: Optional[str] = None, stake_currency: Optional[str] = None,
                         candletype: Optional[CandleType] = None, config=Depends(get_config)):

    dh = get_datahandler(config['datadir'], config.get('dataformat_ohlcv'))
    trading_mode: TradingMode = config.get('trading_mode', TradingMode.SPOT)
    pair_interval = dh.ohlcv_get_available_data(config['datadir'], trading_mode)

    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)]
    if candletype:
        pair_interval = [pair for pair in pair_interval if pair[2] == candletype]
    else:
        candle_type = CandleType.get_default(trading_mode)
        pair_interval = [pair for pair in pair_interval if pair[2] == candle_type]

    pair_interval = sorted(pair_interval, key=lambda x: x[0])

    pairs = list({x[0] for x in pair_interval})
    pairs.sort()
    result = {
        'length': len(pairs),
        'pairs': pairs,
        'pair_interval': pair_interval,
    }
    return result


@router.get('/sysinfo', response_model=SysInfo, tags=['info'])
def sysinfo():
    return RPC._rpc_sysinfo()


@router.get('/health', response_model=Health, tags=['info'])
def health(rpc: RPC = Depends(get_rpc)):
    return rpc.health()

1	import logging	1✔
2	from copy import deepcopy	1✔
3	from typing import List, Optional	1✔
4
5	from fastapi import APIRouter, Depends, Query	1✔
6	from fastapi.exceptions import HTTPException	1✔
7
8	from freqtrade import __version__	1✔
9	from freqtrade.data.history import get_datahandler	1✔
10	from freqtrade.enums import CandleType, TradingMode	1✔
11	from freqtrade.exceptions import OperationalException	1✔
12	from freqtrade.rpc import RPC	1✔
13	from freqtrade.rpc.api_server.api_schemas import (AvailablePairs, Balances, BlacklistPayload,	1✔
14	BlacklistResponse, Count, DailyWeeklyMonthly,
15	DeleteLockRequest, DeleteTrade, Entry,
16	ExchangeListResponse, Exit, ForceEnterPayload,
17	ForceEnterResponse, ForceExitPayload,
18	FreqAIModelListResponse, Health, Locks,
19	LocksPayload, Logs, MixTag, OpenTradeSchema,
20	PairHistory, PerformanceEntry, Ping, PlotConfig,
21	Profit, ResultMsg, ShowConfig, Stats, StatusMsg,
22	StrategyListResponse, StrategyResponse, SysInfo,
23	Version, WhitelistResponse)
24	from freqtrade.rpc.api_server.deps import get_config, get_exchange, get_rpc, get_rpc_optional	1✔
25	from freqtrade.rpc.rpc import RPCException	1✔
26
27
28	logger = logging.getLogger(__name__)	1✔
29
30	# API version
31	# Pre-1.1, no version was provided
32	# Version increments should happen in "small" steps (1.1, 1.12, ...) unless big changes happen.
33	# 1.11: forcebuy and forcesell accept ordertype
34	# 1.12: add blacklist delete endpoint
35	# 1.13: forcebuy supports stake_amount
36	# versions 2.xx -> futures/short branch
37	# 2.14: Add entry/exit orders to trade response
38	# 2.15: Add backtest history endpoints
39	# 2.16: Additional daily metrics
40	# 2.17: Forceentry - leverage, partial force_exit
41	# 2.20: Add websocket endpoints
42	# 2.21: Add new_candle messagetype
43	# 2.22: Add FreqAI to backtesting
44	# 2.23: Allow plot config request in webserver mode
45	# 2.24: Add cancel_open_order endpoint
46	# 2.25: Add several profit values to /status endpoint
47	# 2.26: increase /balance output
48	# 2.27: Add /trades/<id>/reload endpoint
49	# 2.28: Switch reload endpoint to Post
50	# 2.29: Add /exchanges endpoint
51	# 2.30: new /pairlists endpoint
52	# 2.31: new /backtest/history/ delete endpoint
53	# 2.32: new /backtest/history/ patch endpoint
54	# 2.33: Additional weekly/monthly metrics
55	# 2.34: new entries/exits/mix_tags endpoints
56	API_VERSION = 2.34	1✔
57
58	# Public API, requires no auth.
59	router_public = APIRouter()	1✔
60	# Private API, protected by authentication
61	router = APIRouter()	1✔
62
63
64	@router_public.get('/ping', response_model=Ping)	1✔
65	def ping():	1✔
66	"""simple ping"""
67	return {"status": "pong"}	1✔
68
69
70	@router.get('/version', response_model=Version, tags=['info'])	1✔
71	def version():	1✔
72	""" Bot Version info"""
73	return {"version": __version__}	1✔
74
75
76	@router.get('/balance', response_model=Balances, tags=['info'])	1✔
77	def balance(rpc: RPC = Depends(get_rpc), config=Depends(get_config)):	1✔
78	"""Account Balances"""
79	return rpc._rpc_balance(config['stake_currency'], config.get('fiat_display_currency', ''),)	1✔
80
81
82	@router.get('/count', response_model=Count, tags=['info'])	1✔
83	def count(rpc: RPC = Depends(get_rpc)):	1✔
84	return rpc._rpc_count()	1✔
85
86
87	@router.get('/entries', response_model=List[Entry], tags=['info'])	1✔
88	def entries(pair: Optional[str] = None, rpc: RPC = Depends(get_rpc)):	1✔
89	return rpc._rpc_enter_tag_performance(pair)	1✔
90
91
92	@router.get('/exits', response_model=List[Exit], tags=['info'])	1✔
93	def exits(pair: Optional[str] = None, rpc: RPC = Depends(get_rpc)):	1✔
94	return rpc._rpc_exit_reason_performance(pair)	1✔
95
96
97	@router.get('/mix_tags', response_model=List[MixTag], tags=['info'])	1✔
98	def mix_tags(pair: Optional[str] = None, rpc: RPC = Depends(get_rpc)):	1✔
99	return rpc._rpc_mix_tag_performance(pair)	1✔
100
101
102	@router.get('/performance', response_model=List[PerformanceEntry], tags=['info'])	1✔
103	def performance(rpc: RPC = Depends(get_rpc)):	1✔
104	return rpc._rpc_performance()	1✔
105
106
107	@router.get('/profit', response_model=Profit, tags=['info'])	1✔
108	def profit(rpc: RPC = Depends(get_rpc), config=Depends(get_config)):	1✔
109	return rpc._rpc_trade_statistics(config['stake_currency'],	1✔
110	config.get('fiat_display_currency')
111	)
112
113
114	@router.get('/stats', response_model=Stats, tags=['info'])	1✔
115	def stats(rpc: RPC = Depends(get_rpc)):	1✔
116	return rpc._rpc_stats()	1✔
117
118
119	@router.get('/daily', response_model=DailyWeeklyMonthly, tags=['info'])	1✔
120	def daily(timescale: int = 7, rpc: RPC = Depends(get_rpc), config=Depends(get_config)):	1✔
121	return rpc._rpc_timeunit_profit(timescale, config['stake_currency'],	1✔
122	config.get('fiat_display_currency', ''))
123
124
125	@router.get('/weekly', response_model=DailyWeeklyMonthly, tags=['info'])	1✔
126	def weekly(timescale: int = 4, rpc: RPC = Depends(get_rpc), config=Depends(get_config)):	1✔
127	return rpc._rpc_timeunit_profit(timescale, config['stake_currency'],	1✔
128	config.get('fiat_display_currency', ''), 'weeks')
129
130
131	@router.get('/monthly', response_model=DailyWeeklyMonthly, tags=['info'])	1✔
132	def monthly(timescale: int = 3, rpc: RPC = Depends(get_rpc), config=Depends(get_config)):	1✔
133	return rpc._rpc_timeunit_profit(timescale, config['stake_currency'],	1✔
134	config.get('fiat_display_currency', ''), 'months')
135
136
137	@router.get('/status', response_model=List[OpenTradeSchema], tags=['info'])	1✔
138	def status(rpc: RPC = Depends(get_rpc)):	1✔
139	try:	1✔
140	return rpc._rpc_trade_status()	1✔
141	except RPCException:	1✔
142	return []	1✔
143
144
145	# Using the responsemodel here will cause a ~100% increase in response time (from 1s to 2s)
146	# on big databases. Correct response model: response_model=TradeResponse,
147	@router.get('/trades', tags=['info', 'trading'])	1✔
148	def trades(limit: int = 500, offset: int = 0, rpc: RPC = Depends(get_rpc)):	1✔
149	return rpc._rpc_trade_history(limit, offset=offset, order_by_id=True)	1✔
150
151
152	@router.get('/trade/{tradeid}', response_model=OpenTradeSchema, tags=['info', 'trading'])	1✔
153	def trade(tradeid: int = 0, rpc: RPC = Depends(get_rpc)):	1✔
154	try:	1✔
155	return rpc._rpc_trade_status([tradeid])[0]	1✔
156	except (RPCException, KeyError):	1✔
157	raise HTTPException(status_code=404, detail='Trade not found.')	1✔
158
159
160	@router.delete('/trades/{tradeid}', response_model=DeleteTrade, tags=['info', 'trading'])	1✔
161	def trades_delete(tradeid: int, rpc: RPC = Depends(get_rpc)):	1✔
162	return rpc._rpc_delete(tradeid)	1✔
163
164
165	@router.delete('/trades/{tradeid}/open-order', response_model=OpenTradeSchema, tags=['trading'])	1✔
166	def trade_cancel_open_order(tradeid: int, rpc: RPC = Depends(get_rpc)):	1✔
167	rpc._rpc_cancel_open_order(tradeid)	1✔
168	return rpc._rpc_trade_status([tradeid])[0]	1✔
169
170
171	@router.post('/trades/{tradeid}/reload', response_model=OpenTradeSchema, tags=['trading'])	1✔
172	def trade_reload(tradeid: int, rpc: RPC = Depends(get_rpc)):	1✔
173	rpc._rpc_reload_trade_from_exchange(tradeid)	1✔
174	return rpc._rpc_trade_status([tradeid])[0]	1✔
175
176
177	# TODO: Missing response model
178	@router.get('/edge', tags=['info'])	1✔
179	def edge(rpc: RPC = Depends(get_rpc)):	1✔
180	return rpc._rpc_edge()	1✔
181
182
183	@router.get('/show_config', response_model=ShowConfig, tags=['info'])	1✔
184	def show_config(rpc: Optional[RPC] = Depends(get_rpc_optional), config=Depends(get_config)):	1✔
185	state = ''	1✔
186	strategy_version = None	1✔
187	if rpc:	1✔
188	state = rpc._freqtrade.state	1✔
189	strategy_version = rpc._freqtrade.strategy.version()	1✔
190	resp = RPC._rpc_show_config(config, state, strategy_version)	1✔
191	resp['api_version'] = API_VERSION	1✔
192	return resp	1✔
193
194
195	# /forcebuy is deprecated with short addition. use /forceentry instead
196	@router.post('/forceenter', response_model=ForceEnterResponse, tags=['trading'])	1✔
197	@router.post('/forcebuy', response_model=ForceEnterResponse, tags=['trading'])	1✔
198	def force_entry(payload: ForceEnterPayload, rpc: RPC = Depends(get_rpc)):	1✔
199	ordertype = payload.ordertype.value if payload.ordertype else None	1✔
200
201	trade = rpc._rpc_force_entry(payload.pair, payload.price, order_side=payload.side,	1✔
202	order_type=ordertype, stake_amount=payload.stakeamount,
203	enter_tag=payload.entry_tag or 'force_entry',
204	leverage=payload.leverage)
205
206	if trade:	1✔
207	return ForceEnterResponse.model_validate(trade.to_json())	1✔
208	else:
209	return ForceEnterResponse.model_validate(	1✔
210	{"status": f"Error entering {payload.side} trade for pair {payload.pair}."})
211
212
213	# /forcesell is deprecated with short addition. use /forceexit instead
214	@router.post('/forceexit', response_model=ResultMsg, tags=['trading'])	1✔
215	@router.post('/forcesell', response_model=ResultMsg, tags=['trading'])	1✔
216	def forceexit(payload: ForceExitPayload, rpc: RPC = Depends(get_rpc)):	1✔
217	ordertype = payload.ordertype.value if payload.ordertype else None	1✔
218	return rpc._rpc_force_exit(str(payload.tradeid), ordertype, amount=payload.amount)	1✔
219
220
221	@router.get('/blacklist', response_model=BlacklistResponse, tags=['info', 'pairlist'])	1✔
222	def blacklist(rpc: RPC = Depends(get_rpc)):	1✔
223	return rpc._rpc_blacklist()	1✔
224
225
226	@router.post('/blacklist', response_model=BlacklistResponse, tags=['info', 'pairlist'])	1✔
227	def blacklist_post(payload: BlacklistPayload, rpc: RPC = Depends(get_rpc)):	1✔
228	return rpc._rpc_blacklist(payload.blacklist)	1✔
229
230
231	@router.delete('/blacklist', response_model=BlacklistResponse, tags=['info', 'pairlist'])	1✔
232	def blacklist_delete(pairs_to_delete: List[str] = Query([]), rpc: RPC = Depends(get_rpc)):	1✔
233	"""Provide a list of pairs to delete from the blacklist"""
234
235	return rpc._rpc_blacklist_delete(pairs_to_delete)	1✔
236
237
238	@router.get('/whitelist', response_model=WhitelistResponse, tags=['info', 'pairlist'])	1✔
239	def whitelist(rpc: RPC = Depends(get_rpc)):	1✔
240	return rpc._rpc_whitelist()	1✔
241
242
243	@router.get('/locks', response_model=Locks, tags=['info', 'locks'])	1✔
244	def locks(rpc: RPC = Depends(get_rpc)):	1✔
245	return rpc._rpc_locks()	1✔
246
247
248	@router.delete('/locks/{lockid}', response_model=Locks, tags=['info', 'locks'])	1✔
249	def delete_lock(lockid: int, rpc: RPC = Depends(get_rpc)):	1✔
250	return rpc._rpc_delete_lock(lockid=lockid)	1✔
251
252
253	@router.post('/locks/delete', response_model=Locks, tags=['info', 'locks'])	1✔
254	def delete_lock_pair(payload: DeleteLockRequest, rpc: RPC = Depends(get_rpc)):	1✔
255	return rpc._rpc_delete_lock(lockid=payload.lockid, pair=payload.pair)	1✔
256
257
258	@router.post('/locks', response_model=Locks, tags=['info', 'locks'])	1✔
259	def add_locks(payload: List[LocksPayload], rpc: RPC = Depends(get_rpc)):	1✔
260	for lock in payload:	1✔
261	rpc._rpc_add_lock(lock.pair, lock.until, lock.reason, lock.side)	1✔
262	return rpc._rpc_locks()	1✔
263
264
265	@router.get('/logs', response_model=Logs, tags=['info'])	1✔
266	def logs(limit: Optional[int] = None):	1✔
267	return RPC._rpc_get_logs(limit)	1✔
268
269
270	@router.post('/start', response_model=StatusMsg, tags=['botcontrol'])	1✔
271	def start(rpc: RPC = Depends(get_rpc)):	1✔
272	return rpc._rpc_start()	1✔
273
274
275	@router.post('/stop', response_model=StatusMsg, tags=['botcontrol'])	1✔
276	def stop(rpc: RPC = Depends(get_rpc)):	1✔
277	return rpc._rpc_stop()	1✔
278
279
280	@router.post('/stopentry', response_model=StatusMsg, tags=['botcontrol'])	1✔
281	@router.post('/stopbuy', response_model=StatusMsg, tags=['botcontrol'])	1✔
282	def stop_buy(rpc: RPC = Depends(get_rpc)):	1✔
283	return rpc._rpc_stopentry()	1✔
284
285
286	@router.post('/reload_config', response_model=StatusMsg, tags=['botcontrol'])	1✔
287	def reload_config(rpc: RPC = Depends(get_rpc)):	1✔
288	return rpc._rpc_reload_config()	1✔
289
290
291	@router.get('/pair_candles', response_model=PairHistory, tags=['candle data'])	1✔
292	def pair_candles(	1✔
293	pair: str, timeframe: str, limit: Optional[int] = None, rpc: RPC = Depends(get_rpc)):
294	return rpc._rpc_analysed_dataframe(pair, timeframe, limit)	1✔
295
296
297	@router.get('/pair_history', response_model=PairHistory, tags=['candle data'])	1✔
298	def pair_history(pair: str, timeframe: str, timerange: str, strategy: str,	1✔
299	freqaimodel: Optional[str] = None,
300	config=Depends(get_config), exchange=Depends(get_exchange)):
301	# The initial call to this endpoint can be slow, as it may need to initialize
302	# the exchange class.
303	config = deepcopy(config)	1✔
304	config.update({	1✔
305	'strategy': strategy,
306	'timerange': timerange,
307	'freqaimodel': freqaimodel if freqaimodel else config.get('freqaimodel'),
308	})
309	try:	1✔
310	return RPC._rpc_analysed_history_full(config, pair, timeframe, exchange)	1✔
311	except Exception as e:	1✔
312	raise HTTPException(status_code=502, detail=str(e))	1✔
313
314
315	@router.get('/plot_config', response_model=PlotConfig, tags=['candle data'])	1✔
316	def plot_config(strategy: Optional[str] = None, config=Depends(get_config),	1✔
317	rpc: Optional[RPC] = Depends(get_rpc_optional)):
318	if not strategy:	1✔
319	if not rpc:	1✔
320	raise RPCException("Strategy is mandatory in webserver mode.")	×
321	return PlotConfig.model_validate(rpc._rpc_plot_config())	1✔
322	else:
323	config1 = deepcopy(config)	1✔
324	config1.update({	1✔
325	'strategy': strategy
326	})
327	try:	1✔
328	return PlotConfig.model_validate(RPC._rpc_plot_config_with_strategy(config1))	1✔
329	except Exception as e:	1✔
330	raise HTTPException(status_code=502, detail=str(e))	1✔
331
332
333	@router.get('/strategies', response_model=StrategyListResponse, tags=['strategy'])	1✔
334	def list_strategies(config=Depends(get_config)):	1✔
335	from freqtrade.resolvers.strategy_resolver import StrategyResolver	1✔
336	strategies = StrategyResolver.search_all_objects(	1✔
337	config, False, config.get('recursive_strategy_search', False))
338	strategies = sorted(strategies, key=lambda x: x['name'])	1✔
339
340	return {'strategies': [x['name'] for x in strategies]}	1✔
341
342
343	@router.get('/strategy/{strategy}', response_model=StrategyResponse, tags=['strategy'])	1✔
344	def get_strategy(strategy: str, config=Depends(get_config)):	1✔
345	if ":" in strategy:	1✔
346	raise HTTPException(status_code=500, detail="base64 encoded strategies are not allowed.")	1✔
347
348	config_ = deepcopy(config)	1✔
349	from freqtrade.resolvers.strategy_resolver import StrategyResolver	1✔
350	try:	1✔
351	strategy_obj = StrategyResolver._load_strategy(strategy, config_,	1✔
352	extra_dir=config_.get('strategy_path'))
353	except OperationalException:	1✔
354	raise HTTPException(status_code=404, detail='Strategy not found')	1✔
355	except Exception as e:	×
356	raise HTTPException(status_code=502, detail=str(e))	×
357	return {	1✔
358	'strategy': strategy_obj.get_strategy_name(),
359	'code': strategy_obj.__source__,
360	'timeframe': getattr(strategy_obj, 'timeframe', None),
361	}
362
363
364	@router.get('/exchanges', response_model=ExchangeListResponse, tags=[])	1✔
365	def list_exchanges(config=Depends(get_config)):	1✔
366	from freqtrade.exchange import list_available_exchanges	1✔
367	exchanges = list_available_exchanges(config)	1✔
368	return {	1✔
369	'exchanges': exchanges,
370	}
371
372
373	@router.get('/freqaimodels', response_model=FreqAIModelListResponse, tags=['freqai'])	1✔
374	def list_freqaimodels(config=Depends(get_config)):	1✔
375	from freqtrade.resolvers.freqaimodel_resolver import FreqaiModelResolver	1✔
376	models = FreqaiModelResolver.search_all_objects(	1✔
377	config, False)
378	models = sorted(models, key=lambda x: x['name'])	1✔
379
380	return {'freqaimodels': [x['name'] for x in models]}	1✔
381
382
383	@router.get('/available_pairs', response_model=AvailablePairs, tags=['candle data'])	1✔
384	def list_available_pairs(timeframe: Optional[str] = None, stake_currency: Optional[str] = None,	1✔
385	candletype: Optional[CandleType] = None, config=Depends(get_config)):
386
387	dh = get_datahandler(config['datadir'], config.get('dataformat_ohlcv'))	1✔
388	trading_mode: TradingMode = config.get('trading_mode', TradingMode.SPOT)	1✔
389	pair_interval = dh.ohlcv_get_available_data(config['datadir'], trading_mode)	1✔
390
391	if timeframe:	1✔
392	pair_interval = [pair for pair in pair_interval if pair[1] == timeframe]	1✔
393	if stake_currency:	1✔
394	pair_interval = [pair for pair in pair_interval if pair[0].endswith(stake_currency)]	1✔
395	if candletype:	1✔
396	pair_interval = [pair for pair in pair_interval if pair[2] == candletype]	1✔
397	else:
398	candle_type = CandleType.get_default(trading_mode)	1✔
399	pair_interval = [pair for pair in pair_interval if pair[2] == candle_type]	1✔
400
401	pair_interval = sorted(pair_interval, key=lambda x: x[0])	1✔
402
403	pairs = list({x[0] for x in pair_interval})	1✔
404	pairs.sort()	1✔
405	result = {	1✔
406	'length': len(pairs),
407	'pairs': pairs,
408	'pair_interval': pair_interval,
409	}
410	return result	1✔
411
412
413	@router.get('/sysinfo', response_model=SysInfo, tags=['info'])	1✔
414	def sysinfo():	1✔
415	return RPC._rpc_sysinfo()	1✔
416
417
418	@router.get('/health', response_model=Health, tags=['info'])	1✔
419	def health(rpc: RPC = Depends(get_rpc)):	1✔
420	return rpc.health()	1✔

freqtrade / freqtrade / 9394559170

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous