4131164979

Build Type

push

github-actions

Committed by Matthias

Commit Message

filled-date shouldn't update again

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

17024 of 17946 relevant lines covered (94.86%)

0.95 hits per line

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

99.46

/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, Daily,
                                                  DeleteLockRequest, DeleteTrade, ForceEnterPayload,
                                                  ForceEnterResponse, ForceExitPayload,
                                                  FreqAIModelListResponse, Health, Locks, Logs,
                                                  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
API_VERSION = 2.23

# 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('/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=Daily, 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('/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)


# 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.parse_obj(trade.to_json())
    else:
        return ForceEnterResponse.parse_obj(
            {"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(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.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,
                 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,
    })
    return RPC._rpc_analysed_history_full(config, pair, timeframe, timerange, exchange)


@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.parse_obj(rpc._rpc_plot_config())
    else:
        config1 = deepcopy(config)
        config1.update({
            'strategy': strategy
        })
        return PlotConfig.parse_obj(RPC._rpc_plot_config_with_strategy(config1))


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

    return {
        'strategy': strategy_obj.get_strategy_name(),
        'code': strategy_obj.__source__,
    }


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

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


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

freqtrade / freqtrade / 4131164979

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