adelorenzo/stable
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #4138 from freqtrade/fastapi

Browse Source 
use Fastapi instead of flask for API operations
This commit is contained in:
Matthias 2021-01-05 10:07:19 +01:00 committed by GitHub
commit c010cdf894
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 1250 additions and 1024 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
config.json.example
View File

@ -79,11 +79,11 @@
"enabled": false,
"listen_ip_address": "127.0.0.1",
"listen_port": 8080,
"verbosity": "info",
"verbosity": "error",
"jwt_secret_key": "somethingrandom",
"CORS_origins": [],
"username": "",
"password": ""
"username": "freqtrader",
"password": "SuperSecurePassword"
},
"initial_state": "running",
"forcebuy_enable": false,

6
config_binance.json.example
View File

@ -84,11 +84,11 @@
"enabled": false,
"listen_ip_address": "127.0.0.1",
"listen_port": 8080,
"verbosity": "info",
"verbosity": "error",
"jwt_secret_key": "somethingrandom",
"CORS_origins": [],
"username": "",
"password": ""
"username": "freqtrader",
"password": "SuperSecurePassword"
},
"initial_state": "running",
"forcebuy_enable": false,

3
config_full.json.example
View File

@ -164,7 +164,8 @@
"enabled": false,
"listen_ip_address": "127.0.0.1",
"listen_port": 8080,
"verbosity": "info",
"verbosity": "error",
"enable_openapi": false,
"jwt_secret_key": "somethingrandom",
"CORS_origins": [],
"username": "freqtrader",

6
config_kraken.json.example
View File

@ -89,11 +89,11 @@
"enabled": false,
"listen_ip_address": "127.0.0.1",
"listen_port": 8080,
"verbosity": "info",
"verbosity": "error",
"jwt_secret_key": "somethingrandom",
"CORS_origins": [],
"username": "",
"password": ""
"username": "freqtrader",
"password": "SuperSecurePassword"
},
"initial_state": "running",
"forcebuy_enable": false,

8
docs/rest-api.md
View File

@ -11,7 +11,8 @@ Sample configuration:
"enabled": true,
"listen_ip_address": "127.0.0.1",
"listen_port": 8080,
"verbosity": "info",
"verbosity": "error",
"enable_openapi": false,
"jwt_secret_key": "somethingrandom",
"CORS_origins": [],
"username": "Freqtrader",
@ -263,6 +264,11 @@ whitelist
```
## OpenAPI interface
To enable the builtin openAPI interface, specify `"enable_openapi": true` in the api_server configuration.
This will enable the Swagger UI at the `/docs` endpoint. By default, that's running at http://localhost:8080/docs/ - but it'll depend on your settings.
## Advanced API usage using JWT tokens
!!! Note

665
freqtrade/rpc/api_server.py
View File

@ -1,665 +0,0 @@
import logging
import threading
from copy import deepcopy
from datetime import date, datetime
from ipaddress import IPv4Address
from pathlib import Path
from typing import Any, Callable, Dict
from arrow import Arrow
from flask import Flask, jsonify, request
from flask.json import JSONEncoder
from flask_cors import CORS
from flask_jwt_extended import (JWTManager, create_access_token, create_refresh_token,
get_jwt_identity, jwt_refresh_token_required,
verify_jwt_in_request_optional)
from werkzeug.security import safe_str_cmp
from werkzeug.serving import make_server
from freqtrade.__init__ import __version__
from freqtrade.constants import DATETIME_PRINT_FORMAT, USERPATH_STRATEGIES
from freqtrade.exceptions import OperationalException
from freqtrade.persistence import Trade
from freqtrade.rpc.rpc import RPC, RPCException, RPCHandler
logger = logging.getLogger(__name__)
BASE_URI = "/api/v1"
class FTJSONEncoder(JSONEncoder):
def default(self, obj):
try:
if isinstance(obj, Arrow):
return obj.for_json()
elif isinstance(obj, datetime):
return obj.strftime(DATETIME_PRINT_FORMAT)
elif isinstance(obj, date):
return obj.strftime("%Y-%m-%d")
iterable = iter(obj)
except TypeError:
pass
else:
return list(iterable)
return JSONEncoder.default(self, obj)
# Type should really be Callable[[ApiServer, Any], Any], but that will create a circular dependency
def require_login(func: Callable[[Any, Any], Any]):
def func_wrapper(obj, *args, **kwargs):
verify_jwt_in_request_optional()
auth = request.authorization
if get_jwt_identity() or auth and obj.check_auth(auth.username, auth.password):
return func(obj, *args, **kwargs)
else:
return jsonify({"error": "Unauthorized"}), 401
return func_wrapper
# Type should really be Callable[[ApiServer], Any], but that will create a circular dependency
def rpc_catch_errors(func: Callable[..., Any]):
def func_wrapper(obj, *args, **kwargs):
try:
return func(obj, *args, **kwargs)
except RPCException as e:
logger.exception("API Error calling %s: %s", func.__name__, e)
return obj.rest_error(f"Error querying {func.__name__}: {e}")
return func_wrapper
def shutdown_session(exception=None):
# Remove scoped session
Trade.session.remove()
class ApiServer(RPCHandler):
"""
This class runs api server and provides rpc.rpc functionality to it
This class starts a non-blocking thread the api server runs within
"""
def check_auth(self, username, password):
return (safe_str_cmp(username, self._config['api_server'].get('username')) and
safe_str_cmp(password, self._config['api_server'].get('password')))
def __init__(self, rpc: RPC, config: Dict[str, Any]) -> None:
"""
Init the api server, and init the super class RPCHandler
:param rpc: instance of RPC Helper class
:param config: Configuration object
:return: None
"""
super().__init__(rpc, config)
self.app = Flask(__name__)
self._cors = CORS(self.app,
resources={r"/api/*": {
"supports_credentials": True,
"origins": self._config['api_server'].get('CORS_origins', [])}}
)
# Setup the Flask-JWT-Extended extension
self.app.config['JWT_SECRET_KEY'] = self._config['api_server'].get(
'jwt_secret_key', 'super-secret')
self.jwt = JWTManager(self.app)
self.app.json_encoder = FTJSONEncoder
self.app.teardown_appcontext(shutdown_session)
# Register application handling
self.register_rest_rpc_urls()
thread = threading.Thread(target=self.run, daemon=True)
thread.start()
def cleanup(self) -> None:
logger.info("Stopping API Server")
self.srv.shutdown()
def run(self):
"""
Method that runs flask app in its own thread forever.
Section to handle configuration and running of the Rest server
also to check and warn if not bound to a loopback, warn on security risk.
"""
rest_ip = self._config['api_server']['listen_ip_address']
rest_port = self._config['api_server']['listen_port']
logger.info(f'Starting HTTP Server at {rest_ip}:{rest_port}')
if not IPv4Address(rest_ip).is_loopback:
logger.warning("SECURITY WARNING - Local Rest Server listening to external connections")
logger.warning("SECURITY WARNING - This is insecure please set to your loopback,"
"e.g 127.0.0.1 in config.json")
if not self._config['api_server'].get('password'):
logger.warning("SECURITY WARNING - No password for local REST Server defined. "
"Please make sure that this is intentional!")
# Run the Server
logger.info('Starting Local Rest Server.')
try:
self.srv = make_server(rest_ip, rest_port, self.app)
self.srv.serve_forever()
except Exception:
logger.exception("Api server failed to start.")
logger.info('Local Rest Server started.')
def send_msg(self, msg: Dict[str, str]) -> None:
"""
We don't push to endpoints at the moment.
Take a look at webhooks for that functionality.
"""
pass
def rest_error(self, error_msg, error_code=502):
return jsonify({"error": error_msg}), error_code
def register_rest_rpc_urls(self):
"""
Registers flask app URLs that are calls to functionality in rpc.rpc.
First two arguments passed are /URL and 'Label'
Label can be used as a shortcut when refactoring
:return:
"""
self.app.register_error_handler(404, self.page_not_found)
# Actions to control the bot
self.app.add_url_rule(f'{BASE_URI}/token/login', 'login',
view_func=self._token_login, methods=['POST'])
self.app.add_url_rule(f'{BASE_URI}/token/refresh', 'token_refresh',
view_func=self._token_refresh, methods=['POST'])
self.app.add_url_rule(f'{BASE_URI}/start', 'start',
view_func=self._start, methods=['POST'])
self.app.add_url_rule(f'{BASE_URI}/stop', 'stop', view_func=self._stop, methods=['POST'])
self.app.add_url_rule(f'{BASE_URI}/stopbuy', 'stopbuy',
view_func=self._stopbuy, methods=['POST'])
self.app.add_url_rule(f'{BASE_URI}/reload_config', 'reload_config',
view_func=self._reload_config, methods=['POST'])
# Info commands
self.app.add_url_rule(f'{BASE_URI}/balance', 'balance',
view_func=self._balance, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/count', 'count', view_func=self._count, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/locks', 'locks', view_func=self._locks, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/daily', 'daily', view_func=self._daily, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/edge', 'edge', view_func=self._edge, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/logs', 'log', view_func=self._get_logs, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/profit', 'profit',
view_func=self._profit, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/stats', 'stats',
view_func=self._stats, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/performance', 'performance',
view_func=self._performance, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/status', 'status',
view_func=self._status, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/version', 'version',
view_func=self._version, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/show_config', 'show_config',
view_func=self._show_config, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/ping', 'ping',
view_func=self._ping, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/trades', 'trades',
view_func=self._trades, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/trades/<int:tradeid>', 'trades_delete',
view_func=self._trades_delete, methods=['DELETE'])
self.app.add_url_rule(f'{BASE_URI}/pair_candles', 'pair_candles',
view_func=self._analysed_candles, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/pair_history', 'pair_history',
view_func=self._analysed_history, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/plot_config', 'plot_config',
view_func=self._plot_config, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/strategies', 'strategies',
view_func=self._list_strategies, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/strategy/<string:strategy>', 'strategy',
view_func=self._get_strategy, methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/available_pairs', 'pairs',
view_func=self._list_available_pairs, methods=['GET'])
# Combined actions and infos
self.app.add_url_rule(f'{BASE_URI}/blacklist', 'blacklist', view_func=self._blacklist,
methods=['GET', 'POST'])
self.app.add_url_rule(f'{BASE_URI}/whitelist', 'whitelist', view_func=self._whitelist,
methods=['GET'])
self.app.add_url_rule(f'{BASE_URI}/forcebuy', 'forcebuy',
view_func=self._forcebuy, methods=['POST'])
self.app.add_url_rule(f'{BASE_URI}/forcesell', 'forcesell', view_func=self._forcesell,
methods=['POST'])
@require_login
def page_not_found(self, error):
"""
Return "404 not found", 404.
"""
return jsonify({
'status': 'error',
'reason': f"There's no API call for {request.base_url}.",
'code': 404
}), 404
@require_login
@rpc_catch_errors
def _token_login(self):
"""
Handler for /token/login
Returns a JWT token
"""
auth = request.authorization
if auth and self.check_auth(auth.username, auth.password):
keystuff = {'u': auth.username}
ret = {
'access_token': create_access_token(identity=keystuff),
'refresh_token': create_refresh_token(identity=keystuff),
}
return jsonify(ret)
return jsonify({"error": "Unauthorized"}), 401
@jwt_refresh_token_required
@rpc_catch_errors
def _token_refresh(self):
"""
Handler for /token/refresh
Returns a JWT token based on a JWT refresh token
"""
current_user = get_jwt_identity()
new_token = create_access_token(identity=current_user, fresh=False)
ret = {'access_token': new_token}
return jsonify(ret)
@require_login
@rpc_catch_errors
def _start(self):
"""
Handler for /start.
Starts TradeThread in bot if stopped.
"""
msg = self._rpc._rpc_start()
return jsonify(msg)
@require_login
@rpc_catch_errors
def _stop(self):
"""
Handler for /stop.
Stops TradeThread in bot if running
"""
msg = self._rpc._rpc_stop()
return jsonify(msg)
@require_login
@rpc_catch_errors
def _stopbuy(self):
"""
Handler for /stopbuy.
Sets max_open_trades to 0 and gracefully sells all open trades
"""
msg = self._rpc._rpc_stopbuy()
return jsonify(msg)
@rpc_catch_errors
def _ping(self):
"""
simple ping version
"""
return jsonify({"status": "pong"})
@require_login
@rpc_catch_errors
def _version(self):
"""
Prints the bot's version
"""
return jsonify({"version": __version__})
@require_login
@rpc_catch_errors
def _show_config(self):
"""
Prints the bot's version
"""
return jsonify(RPC._rpc_show_config(self._config, self._rpc._freqtrade.state))
@require_login
@rpc_catch_errors
def _reload_config(self):
"""
Handler for /reload_config.
Triggers a config file reload
"""
msg = self._rpc._rpc_reload_config()
return jsonify(msg)
@require_login
@rpc_catch_errors
def _count(self):
"""
Handler for /count.
Returns the number of trades running
"""
msg = self._rpc._rpc_count()
return jsonify(msg)
@require_login
@rpc_catch_errors
def _locks(self):
"""
Handler for /locks.
Returns the currently active locks.
"""
return jsonify(self._rpc._rpc_locks())
@require_login
@rpc_catch_errors
def _daily(self):
"""
Returns the last X days trading stats summary.
:return: stats
"""
timescale = request.args.get('timescale', 7)
timescale = int(timescale)
stats = self._rpc._rpc_daily_profit(timescale,
self._config['stake_currency'],
self._config.get('fiat_display_currency', '')
)
return jsonify(stats)
@require_login
@rpc_catch_errors
def _get_logs(self):
"""
Returns latest logs
get:
param:
limit: Only get a certain number of records
"""
limit = int(request.args.get('limit', 0)) or None
return jsonify(RPC._rpc_get_logs(limit))
@require_login
@rpc_catch_errors
def _edge(self):
"""
Returns information related to Edge.
:return: edge stats
"""
stats = self._rpc._rpc_edge()
return jsonify(stats)
@require_login
@rpc_catch_errors
def _profit(self):
"""
Handler for /profit.
Returns a cumulative profit statistics
:return: stats
"""
stats = self._rpc._rpc_trade_statistics(self._config['stake_currency'],
self._config.get('fiat_display_currency')
)
return jsonify(stats)
@require_login
@rpc_catch_errors
def _stats(self):
"""
Handler for /stats.
Returns a Object with "durations" and "sell_reasons" as keys.
"""
stats = self._rpc._rpc_stats()
return jsonify(stats)
@require_login
@rpc_catch_errors
def _performance(self):
"""
Handler for /performance.
Returns a cumulative performance statistics
:return: stats
"""
stats = self._rpc._rpc_performance()
return jsonify(stats)
@require_login
@rpc_catch_errors
def _status(self):
"""
Handler for /status.
Returns the current status of the trades in json format
"""
try:
results = self._rpc._rpc_trade_status()
return jsonify(results)
except RPCException:
return jsonify([])
@require_login
@rpc_catch_errors
def _balance(self):
"""
Handler for /balance.
Returns the current status of the trades in json format
"""
results = self._rpc._rpc_balance(self._config['stake_currency'],
self._config.get('fiat_display_currency', ''))
return jsonify(results)
@require_login
@rpc_catch_errors
def _trades(self):
"""
Handler for /trades.
Returns the X last trades in json format
"""
limit = int(request.args.get('limit', 0))
results = self._rpc._rpc_trade_history(limit)
return jsonify(results)
@require_login
@rpc_catch_errors
def _trades_delete(self, tradeid: int):
"""
Handler for DELETE /trades/<tradeid> endpoint.
Removes the trade from the database (tries to cancel open orders first!)
get:
param:
tradeid: Numeric trade-id assigned to the trade.
"""
result = self._rpc._rpc_delete(tradeid)
return jsonify(result)
@require_login
@rpc_catch_errors
def _whitelist(self):
"""
Handler for /whitelist.
"""
results = self._rpc._rpc_whitelist()
return jsonify(results)
@require_login
@rpc_catch_errors
def _blacklist(self):
"""
Handler for /blacklist.
"""
add = request.json.get("blacklist", None) if request.method == 'POST' else None
results = self._rpc._rpc_blacklist(add)
return jsonify(results)
@require_login
@rpc_catch_errors
def _forcebuy(self):
"""
Handler for /forcebuy.
"""
asset = request.json.get("pair")
price = request.json.get("price", None)
price = float(price) if price is not None else price
trade = self._rpc._rpc_forcebuy(asset, price)
if trade:
return jsonify(trade.to_json())
else:
return jsonify({"status": f"Error buying pair {asset}."})
@require_login
@rpc_catch_errors
def _forcesell(self):
"""
Handler for /forcesell.
"""
tradeid = request.json.get("tradeid")
results = self._rpc._rpc_forcesell(tradeid)
return jsonify(results)
@require_login
@rpc_catch_errors
def _analysed_candles(self):
"""
Handler for /pair_candles.
Returns the dataframe the bot is using during live/dry operations.
Takes the following get arguments:
get:
parameters:
- pair: Pair
- timeframe: Timeframe to get data for (should be aligned to strategy.timeframe)
- limit: Limit return length to the latest X candles
"""
pair = request.args.get("pair")
timeframe = request.args.get("timeframe")
limit = request.args.get("limit", type=int)
if not pair or not timeframe:
return self.rest_error("Mandatory parameter missing.", 400)
results = self._rpc._rpc_analysed_dataframe(pair, timeframe, limit)
return jsonify(results)
@require_login
@rpc_catch_errors
def _analysed_history(self):
"""
Handler for /pair_history.
Returns the dataframe of a given timerange
Takes the following get arguments:
get:
parameters:
- pair: Pair
- timeframe: Timeframe to get data for (should be aligned to strategy.timeframe)
- strategy: Strategy to use - Must exist in configured strategy-path!
- timerange: timerange in the format YYYYMMDD-YYYYMMDD (YYYYMMDD- or (-YYYYMMDD))
are als possible. If omitted uses all available data.
"""
pair = request.args.get("pair")
timeframe = request.args.get("timeframe")
timerange = request.args.get("timerange")
strategy = request.args.get("strategy")
if not pair or not timeframe or not timerange or not strategy:
return self.rest_error("Mandatory parameter missing.", 400)
config = deepcopy(self._config)
config.update({
'strategy': strategy,
})
results = RPC._rpc_analysed_history_full(config, pair, timeframe, timerange)
return jsonify(results)
@require_login
@rpc_catch_errors
def _plot_config(self):
"""
Handler for /plot_config.
"""
return jsonify(self._rpc._rpc_plot_config())
@require_login
@rpc_catch_errors
def _list_strategies(self):
directory = Path(self._config.get(
'strategy_path', self._config['user_data_dir'] / USERPATH_STRATEGIES))
from freqtrade.resolvers.strategy_resolver import StrategyResolver
strategy_objs = StrategyResolver.search_all_objects(directory, False)
strategy_objs = sorted(strategy_objs, key=lambda x: x['name'])
return jsonify({'strategies': [x['name'] for x in strategy_objs]})
@require_login
@rpc_catch_errors
def _get_strategy(self, strategy: str):
"""
Get a single strategy
get:
parameters:
- strategy: Only get this strategy
"""
config = deepcopy(self._config)
from freqtrade.resolvers.strategy_resolver import StrategyResolver
try:
strategy_obj = StrategyResolver._load_strategy(strategy, config,
extra_dir=config.get('strategy_path'))
except OperationalException:
return self.rest_error("Strategy not found.", 404)
return jsonify({
'strategy': strategy_obj.get_strategy_name(),
'code': strategy_obj.__source__,
})
@require_login
@rpc_catch_errors
def _list_available_pairs(self):
"""
Handler for /available_pairs.
Returns an object, with pairs, available pair length and pair_interval combinations
Takes the following get arguments:
get:
parameters:
- stake_currency: Filter on this stake currency
- timeframe: Timeframe to get data for Filter elements to this timeframe
"""
timeframe = request.args.get("timeframe")
stake_currency = request.args.get("stake_currency")
from freqtrade.data.history import get_datahandler
dh = get_datahandler(self._config['datadir'], self._config.get('dataformat_ohlcv', None))
pair_interval = dh.ohlcv_get_available_data(self._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 jsonify(result)

2
freqtrade/rpc/api_server/__init__.py Normal file
View File

@ -0,0 +1,2 @@
# flake8: noqa: F401
from .webserver import ApiServer

106
freqtrade/rpc/api_server/api_auth.py Normal file
View File

@ -0,0 +1,106 @@
import secrets
from datetime import datetime, timedelta
import jwt
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.security.http import HTTPBasic, HTTPBasicCredentials
from freqtrade.rpc.api_server.api_schemas import AccessAndRefreshToken, AccessToken
from freqtrade.rpc.api_server.deps import get_api_config
ALGORITHM = "HS256"
router_login = APIRouter()
def verify_auth(api_config, username: str, password: str):
"""Verify username/password"""
return (secrets.compare_digest(username, api_config.get('username')) and
secrets.compare_digest(password, api_config.get('password')))
httpbasic = HTTPBasic(auto_error=False)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)
def get_user_from_token(token, secret_key: str, token_type: str = "access"):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, secret_key, algorithms=[ALGORITHM])
username: str = payload.get("identity", {}).get('u')
if username is None:
raise credentials_exception
if payload.get("type") != token_type:
raise credentials_exception
except jwt.PyJWTError:
raise credentials_exception
return username
def create_token(data: dict, secret_key: str, token_type: str = "access") -> bytes:
to_encode = data.copy()
if token_type == "access":
expire = datetime.utcnow() + timedelta(minutes=15)
elif token_type == "refresh":
expire = datetime.utcnow() + timedelta(days=30)
else:
raise ValueError()
to_encode.update({
"exp": expire,
"iat": datetime.utcnow(),
"type": token_type,
})
encoded_jwt = jwt.encode(to_encode, secret_key, algorithm=ALGORITHM)
return encoded_jwt
def http_basic_or_jwt_token(form_data: HTTPBasicCredentials = Depends(httpbasic),
token: str = Depends(oauth2_scheme),
api_config=Depends(get_api_config)):
if token:
return get_user_from_token(token, api_config.get('jwt_secret_key', 'super-secret'))
elif form_data and verify_auth(api_config, form_data.username, form_data.password):
return form_data.username
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Unauthorized",
)
@router_login.post('/token/login', response_model=AccessAndRefreshToken)
def token_login(form_data: HTTPBasicCredentials = Depends(HTTPBasic()),
api_config=Depends(get_api_config)):
if verify_auth(api_config, form_data.username, form_data.password):
token_data = {'identity': {'u': form_data.username}}
access_token = create_token(token_data, api_config.get('jwt_secret_key', 'super-secret'))
refresh_token = create_token(token_data, api_config.get('jwt_secret_key', 'super-secret'),
token_type="refresh")
return {
"access_token": access_token,
"refresh_token": refresh_token,
}
else:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
)
@router_login.post('/token/refresh', response_model=AccessToken)
def token_refresh(token: str = Depends(oauth2_scheme), api_config=Depends(get_api_config)):
# Refresh token
u = get_user_from_token(token, api_config.get(
'jwt_secret_key', 'super-secret'), 'refresh')
token_data = {'identity': {'u': u}}
access_token = create_token(token_data, api_config.get('jwt_secret_key', 'super-secret'),
token_type="access")
return {'access_token': access_token}

306
freqtrade/rpc/api_server/api_schemas.py Normal file
View File

@ -0,0 +1,306 @@
from datetime import date, datetime
from typing import Any, Dict, List, Optional, TypeVar, Union
from pydantic import BaseModel
from freqtrade.constants import DATETIME_PRINT_FORMAT
class Ping(BaseModel):
status: str
class AccessToken(BaseModel):
access_token: str
class AccessAndRefreshToken(AccessToken):
refresh_token: str
class Version(BaseModel):
version: str
class StatusMsg(BaseModel):
status: str
class ResultMsg(BaseModel):
result: str
class Balance(BaseModel):
currency: str
free: float
balance: float
used: float
est_stake: float
stake: str
class Balances(BaseModel):
currencies: List[Balance]
total: float
symbol: str
value: float
stake: str
note: str
class Count(BaseModel):
current: int
max: int
total_stake: float
class PerformanceEntry(BaseModel):
pair: str
profit: float
count: int
class Profit(BaseModel):
profit_closed_coin: float
profit_closed_percent: float
profit_closed_percent_mean: float
profit_closed_ratio_mean: float
profit_closed_percent_sum: float
profit_closed_ratio_sum: float
profit_closed_fiat: float
profit_all_coin: float
profit_all_percent: float
profit_all_percent_mean: float
profit_all_ratio_mean: float
profit_all_percent_sum: float
profit_all_ratio_sum: float
profit_all_fiat: float
trade_count: int
closed_trade_count: int
first_trade_date: str
first_trade_timestamp: int
latest_trade_date: str
latest_trade_timestamp: int
avg_duration: str
best_pair: str
best_rate: float
winning_trades: int
losing_trades: int
class SellReason(BaseModel):
wins: int
losses: int
draws: int
class Stats(BaseModel):
sell_reasons: Dict[str, SellReason]
durations: Dict[str, Union[str, float]]
class DailyRecord(BaseModel):
date: date
abs_profit: float
fiat_value: float
trade_count: int
class Daily(BaseModel):
data: List[DailyRecord]
fiat_display_currency: str
stake_currency: str
class ShowConfig(BaseModel):
dry_run: str
stake_currency: str
stake_amount: Union[float, str]
max_open_trades: int
minimal_roi: Dict[str, Any]
stoploss: float
trailing_stop: bool
trailing_stop_positive: Optional[float]
trailing_stop_positive_offset: Optional[float]
trailing_only_offset_is_reached: Optional[bool]
timeframe: str
timeframe_ms: int
timeframe_min: int
exchange: str
strategy: str
forcebuy_enabled: bool
ask_strategy: Dict[str, Any]
bid_strategy: Dict[str, Any]
state: str
runmode: str
class TradeSchema(BaseModel):
trade_id: int
pair: str
is_open: bool
exchange: str
amount: float
amount_requested: float
stake_amount: float
strategy: str
timeframe: int
fee_open: Optional[float]
fee_open_cost: Optional[float]
fee_open_currency: Optional[str]
fee_close: Optional[float]
fee_close_cost: Optional[float]
fee_close_currency: Optional[str]
open_date_hum: str
open_date: str
open_timestamp: int
open_rate: float
open_rate_requested: Optional[float]
open_trade_value: float
close_date_hum: Optional[str]
close_date: Optional[str]
close_timestamp: Optional[int]
close_rate: Optional[float]
close_rate_requested: Optional[float]
close_profit: Optional[float]
close_profit_pct: Optional[float]
close_profit_abs: Optional[float]
profit_ratio: Optional[float]
profit_pct: Optional[float]
profit_abs: Optional[float]
sell_reason: Optional[str]
sell_order_status: Optional[str]
stop_loss_abs: Optional[float]
stop_loss_ratio: Optional[float]
stop_loss_pct: Optional[float]
stoploss_order_id: Optional[str]
stoploss_last_update: Optional[str]
stoploss_last_update_timestamp: Optional[int]
initial_stop_loss_abs: Optional[float]
initial_stop_loss_ratio: Optional[float]
initial_stop_loss_pct: Optional[float]
min_rate: Optional[float]
max_rate: Optional[float]
open_order_id: Optional[str]
class OpenTradeSchema(TradeSchema):
stoploss_current_dist: Optional[float]
stoploss_current_dist_pct: Optional[float]
stoploss_current_dist_ratio: Optional[float]
stoploss_entry_dist: Optional[float]
stoploss_entry_dist_ratio: Optional[float]
base_currency: str
current_profit: float
current_profit_abs: float
current_profit_pct: float
current_rate: float
open_order: Optional[str]
class TradeResponse(BaseModel):
trades: List[TradeSchema]
trades_count: int
ForceBuyResponse = TypeVar('ForceBuyResponse', TradeSchema, StatusMsg)
class LockModel(BaseModel):
active: bool
lock_end_time: str
lock_end_timestamp: int
lock_time: str
lock_timestamp: int
pair: str
reason: str
class Locks(BaseModel):
lock_count: int
locks: List[LockModel]
class Logs(BaseModel):
log_count: int
logs: List[List]
class ForceBuyPayload(BaseModel):
pair: str
price: Optional[float]
class ForceSellPayload(BaseModel):
tradeid: str
class BlacklistPayload(BaseModel):
blacklist: List[str]
class BlacklistResponse(BaseModel):
blacklist: List[str]
blacklist_expanded: List[str]
errors: Dict
length: int
method: List[str]
class WhitelistResponse(BaseModel):
whitelist: List[str]
length: int
method: List[str]
class DeleteTrade(BaseModel):
cancel_order_count: int
result: str
result_msg: str
trade_id: int
class PlotConfig_(BaseModel):
main_plot: Dict[str, Any]
subplots: Optional[Dict[str, Any]]
PlotConfig = TypeVar('PlotConfig', PlotConfig_, Dict)
class StrategyListResponse(BaseModel):
strategies: List[str]
class StrategyResponse(BaseModel):
strategy: str
code: str
class AvailablePairs(BaseModel):
length: int
pairs: List[str]
pair_interval: List[List[str]]
class PairHistory(BaseModel):
strategy: str
pair: str
timeframe: str
timeframe_ms: int
columns: List[str]
data: List[Any]
length: int
buy_signals: int
sell_signals: int
last_analyzed: datetime
last_analyzed_ts: int
data_start_ts: int
data_start: str
data_stop: str
data_stop_ts: int
class Config:
json_encoders = {
datetime: lambda v: v.strftime(DATETIME_PRINT_FORMAT),
}

238
freqtrade/rpc/api_server/api_v1.py Normal file
View File

@ -0,0 +1,238 @@
from copy import deepcopy
from pathlib import Path
from typing import List, Optional
from fastapi import APIRouter, Depends
from fastapi.exceptions import HTTPException
from freqtrade import __version__
from freqtrade.constants import USERPATH_STRATEGIES
from freqtrade.data.history import get_datahandler
from freqtrade.exceptions import OperationalException
from freqtrade.rpc import RPC
from freqtrade.rpc.api_server.api_schemas import (AvailablePairs, Balances, BlacklistPayload,
BlacklistResponse, Count, Daily, DeleteTrade,
ForceBuyPayload, ForceBuyResponse,
ForceSellPayload, Locks, Logs, OpenTradeSchema,
PairHistory, PerformanceEntry, Ping, PlotConfig,
Profit, ResultMsg, ShowConfig, Stats, StatusMsg,
StrategyListResponse, StrategyResponse,
TradeResponse, Version, WhitelistResponse)
from freqtrade.rpc.api_server.deps import get_config, get_rpc, get_rpc_optional
from freqtrade.rpc.rpc import RPCException
# 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_daily_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 []
@router.get('/trades', response_model=TradeResponse, tags=['info', 'trading'])
def trades(limit: int = 0, rpc: RPC = Depends(get_rpc)):
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()
@router.get('/show_config', response_model=ShowConfig, tags=['info'])
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)
@router.post('/forcebuy', response_model=ForceBuyResponse, 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}."}
@router.post('/forcesell', response_model=ResultMsg, tags=['trading'])
def forcesell(payload: ForceSellPayload, rpc: RPC = Depends(get_rpc)):
return rpc._rpc_forcesell(payload.tradeid)
@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)
@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('/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()
@router.get('/pair_candles', response_model=PairHistory, tags=['candle data'])
def pair_candles(pair: str, timeframe: str, limit: Optional[int], 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)):
config = deepcopy(config)
config.update({
'strategy': strategy,
})
return RPC._rpc_analysed_history_full(config, pair, timeframe, timerange)
@router.get('/plot_config', response_model=PlotConfig, tags=['candle data'])
def plot_config(rpc: RPC = Depends(get_rpc)):
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, 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

27
freqtrade/rpc/api_server/deps.py Normal file
View File

@ -0,0 +1,27 @@
from typing import Any, Dict, Optional
from freqtrade.rpc.rpc import RPC, RPCException
from .webserver import ApiServer
def get_rpc_optional() -> Optional[RPC]:
if ApiServer._has_rpc:
return ApiServer._rpc
return None
def get_rpc() -> Optional[RPC]:
_rpc = get_rpc_optional()
if _rpc:
return _rpc
else:
raise RPCException('Bot is not in the correct state')
def get_config() -> Dict[str, Any]:
return ApiServer._config
def get_api_config() -> Dict[str, Any]:
return ApiServer._config['api_server']

27
freqtrade/rpc/api_server/uvicorn_threaded.py Normal file
View File

@ -0,0 +1,27 @@
import contextlib
import threading
import time
import uvicorn
class UvicornServer(uvicorn.Server):
"""
Multithreaded server - as found in https://github.com/encode/uvicorn/issues/742
"""
def install_signal_handlers(self):
"""
In the parent implementation, this starts the thread, therefore we must patch it away here.
"""
pass
@contextlib.contextmanager
def run_in_thread(self):
self.thread = threading.Thread(target=self.run)
self.thread.start()
while not self.started:
time.sleep(1e-3)
def cleanup(self):
self.should_exit = True
self.thread.join()

116
freqtrade/rpc/api_server/webserver.py Normal file
View File

@ -0,0 +1,116 @@
import logging
from ipaddress import IPv4Address
from typing import Any, Dict
import uvicorn
from fastapi import Depends, FastAPI
from fastapi.middleware.cors import CORSMiddleware
from starlette.responses import JSONResponse
from freqtrade.rpc.api_server.uvicorn_threaded import UvicornServer
from freqtrade.rpc.rpc import RPC, RPCException, RPCHandler
logger = logging.getLogger(__name__)
class ApiServer(RPCHandler):
_rpc: RPC
_has_rpc: bool = False
_config: Dict[str, Any] = {}
def __init__(self, rpc: RPC, config: Dict[str, Any]) -> None:
super().__init__(rpc, config)
self._server = None
ApiServer._rpc = rpc
ApiServer._has_rpc = True
ApiServer._config = config
api_config = self._config['api_server']
self.app = FastAPI(title="Freqtrade API",
openapi_url='openapi.json' if api_config.get(
'enable_openapi', False) else None,
redoc_url=None,
)
self.configure_app(self.app, self._config)
self.start_api()
def cleanup(self) -> None:
""" Cleanup pending module resources """
if self._server:
logger.info("Stopping API Server")
self._server.cleanup()
def send_msg(self, msg: Dict[str, str]) -> None:
pass
def handle_rpc_exception(self, request, exc):
logger.exception(f"API Error calling: {exc}")
return JSONResponse(
status_code=502,
content={'error': f"Error querying {request.url.path}: {exc.message}"}
)
def configure_app(self, app: FastAPI, config):
from freqtrade.rpc.api_server.api_auth import http_basic_or_jwt_token, router_login
from freqtrade.rpc.api_server.api_v1 import router as api_v1
from freqtrade.rpc.api_server.api_v1 import router_public as api_v1_public
app.include_router(api_v1_public, prefix="/api/v1")
app.include_router(api_v1, prefix="/api/v1",
dependencies=[Depends(http_basic_or_jwt_token)],
)
app.include_router(router_login, prefix="/api/v1", tags=["auth"])
app.add_middleware(
CORSMiddleware,
allow_origins=config['api_server'].get('CORS_origins', []),
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.add_exception_handler(RPCException, self.handle_rpc_exception)
def start_api(self):
"""
Start API ... should be run in thread.
"""
rest_ip = self._config['api_server']['listen_ip_address']
rest_port = self._config['api_server']['listen_port']
logger.info(f'Starting HTTP Server at {rest_ip}:{rest_port}')
if not IPv4Address(rest_ip).is_loopback:
logger.warning("SECURITY WARNING - Local Rest Server listening to external connections")
logger.warning("SECURITY WARNING - This is insecure please set to your loopback,"
"e.g 127.0.0.1 in config.json")
if not self._config['api_server'].get('password'):
logger.warning("SECURITY WARNING - No password for local REST Server defined. "
"Please make sure that this is intentional!")
if (self._config['api_server'].get('jwt_secret_key', 'super-secret')
in ('super-secret, somethingrandom')):
logger.warning("SECURITY WARNING - `jwt_secret_key` seems to be default."
"Others may be able to log into your bot.")
logger.info('Starting Local Rest Server.')
verbosity = self._config['api_server'].get('verbosity', 'error')
log_config = uvicorn.config.LOGGING_CONFIG
# Change logging of access logs to stderr
log_config["handlers"]["access"]["stream"] = log_config["handlers"]["default"]["stream"]
uvconfig = uvicorn.Config(self.app,
port=rest_port,
host=rest_ip,
use_colors=False,
log_config=log_config,
access_log=True if verbosity != 'error' else False,
)
try:
self._server = UvicornServer(uvconfig)
self._server.run_in_thread()
except Exception:
logger.exception("Api server failed to start.")

2
freqtrade/rpc/rpc.py
View File

@ -111,7 +111,7 @@ class RPC:
self._fiat_converter = CryptoToFiatConverter()
@staticmethod
def _rpc_show_config(config, botstate: State) -> Dict[str, Any]:
def _rpc_show_config(config, botstate: Union[State, str]) -> Dict[str, Any]:
"""
Return a dict of config options.
Explicitly does NOT return the full config to avoid leakage of sensitive

1
freqtrade/rpc/rpc_manager.py
View File

@ -35,6 +35,7 @@ class RPCManager:
if config.get('api_server', {}).get('enabled', False):
logger.info('Enabling rpc.api_server')
from freqtrade.rpc.api_server import ApiServer
self.registered_modules.append(ApiServer(self._rpc, config))
def cleanup(self) -> None:

8
requirements.txt
View File

@ -27,10 +27,10 @@ python-rapidjson==1.0
# Notify systemd
sdnotify==0.3.2
# Api server
flask==1.1.2
flask-jwt-extended==3.25.0
flask-cors==3.0.9
# API Server
fastapi==0.63.0
uvicorn==0.13.2
pyjwt==1.7.1
# Support for colorized terminal output
colorama==0.4.4

6
tests/conftest_trades.py
View File

@ -32,6 +32,7 @@ def mock_trade_1(fee):
exchange='bittrex',
open_order_id='dry_run_buy_12345',
strategy='DefaultStrategy',
timeframe=5,
)
o = Order.parse_from_ccxt_object(mock_order_1(), 'ETH/BTC', 'buy')
trade.orders.append(o)
@ -84,6 +85,7 @@ def mock_trade_2(fee):
is_open=False,
open_order_id='dry_run_sell_12345',
strategy='DefaultStrategy',
timeframe=5,
sell_reason='sell_signal',
open_date=datetime.now(tz=timezone.utc) - timedelta(minutes=20),
close_date=datetime.now(tz=timezone.utc),
@ -132,6 +134,7 @@ def mock_trade_3(fee):
pair='XRP/BTC',
stake_amount=0.001,
amount=123.0,
amount_requested=123.0,
fee_open=fee.return_value,
fee_close=fee.return_value,
open_rate=0.05,
@ -139,6 +142,8 @@ def mock_trade_3(fee):
close_profit=0.01,
exchange='bittrex',
is_open=False,
strategy='DefaultStrategy',
timeframe=5,
sell_reason='roi',
open_date=datetime.now(tz=timezone.utc) - timedelta(minutes=20),
close_date=datetime.now(tz=timezone.utc),
@ -179,6 +184,7 @@ def mock_trade_4(fee):
exchange='bittrex',
open_order_id='prod_buy_12345',
strategy='DefaultStrategy',
timeframe=5,
)
o = Order.parse_from_ccxt_object(mock_order_4(), 'ETC/BTC', 'buy')
trade.orders.append(o)

2
tests/data/test_btanalysis.py
View File

@ -128,7 +128,7 @@ def test_load_trades_from_db(default_conf, fee, mocker):
if col not in ['index', 'open_at_end']:
assert col in trades.columns
trades = load_trades_from_db(db_url=default_conf['db_url'], strategy='DefaultStrategy')
assert len(trades) == 3
assert len(trades) == 4
trades = load_trades_from_db(db_url=default_conf['db_url'], strategy='NoneStrategy')
assert len(trades) == 0

2
tests/optimize/test_hyperoptloss.py
View File

@ -20,7 +20,7 @@ def test_hyperoptlossresolver(mocker, default_conf) -> None:
hl = ShortTradeDurHyperOptLoss
mocker.patch(
'freqtrade.resolvers.hyperopt_resolver.HyperOptLossResolver.load_object',
MagicMock(return_value=hl)
MagicMock(return_value=hl())
)
default_conf.update({'hyperopt_loss': 'SharpeHyperOptLossDaily'})
x = HyperOptLossResolver.load_hyperoptloss(default_conf)

733
tests/rpc/test_rpc_apiserver.py
View File

File diff suppressed because it is too large Load Diff

4
tests/rpc/test_rpc_manager.py
View File

@ -160,7 +160,7 @@ def test_startupmessages_telegram_enabled(mocker, default_conf, caplog) -> None:
def test_init_apiserver_disabled(mocker, default_conf, caplog) -> None:
caplog.set_level(logging.DEBUG)
run_mock = MagicMock()
mocker.patch('freqtrade.rpc.api_server.ApiServer.run', run_mock)
mocker.patch('freqtrade.rpc.api_server.ApiServer.start_api', run_mock)
default_conf['telegram']['enabled'] = False
rpc_manager = RPCManager(get_patched_freqtradebot(mocker, default_conf))
@ -172,7 +172,7 @@ def test_init_apiserver_disabled(mocker, default_conf, caplog) -> None:
def test_init_apiserver_enabled(mocker, default_conf, caplog) -> None:
caplog.set_level(logging.DEBUG)
run_mock = MagicMock()
mocker.patch('freqtrade.rpc.api_server.ApiServer.run', run_mock)
mocker.patch('freqtrade.rpc.api_server.ApiServer.start_api', run_mock)
default_conf["telegram"]["enabled"] = False
default_conf["api_server"] = {"enabled": True,