mirror of
https://github.com/freqtrade/freqtrade.git
synced 2026-01-31 11:20:24 +00:00
chore: update config to modern typing syntax
This commit is contained in:
@@ -1,9 +1,9 @@
|
||||
from typing import Any, Dict
|
||||
from typing import Any
|
||||
|
||||
from freqtrade.enums import RunMode
|
||||
|
||||
|
||||
def start_webserver(args: Dict[str, Any]) -> None:
|
||||
def start_webserver(args: dict[str, Any]) -> None:
|
||||
"""
|
||||
Main entry point for webserver mode
|
||||
"""
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
# Required json-schema for user specified config
|
||||
from typing import Dict
|
||||
|
||||
from freqtrade.constants import (
|
||||
AVAILABLE_DATAHANDLERS,
|
||||
@@ -23,7 +22,7 @@ from freqtrade.constants import (
|
||||
from freqtrade.enums import RPCMessageType
|
||||
|
||||
|
||||
__MESSAGE_TYPE_DICT: Dict[str, Dict[str, str]] = {x: {"type": "object"} for x in RPCMessageType}
|
||||
__MESSAGE_TYPE_DICT: dict[str, dict[str, str]] = {x: {"type": "object"} for x in RPCMessageType}
|
||||
|
||||
__IN_STRATEGY = "\nUsually specified in the strategy and missing in the configuration."
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import logging
|
||||
from typing import Any, Dict
|
||||
from typing import Any
|
||||
|
||||
from freqtrade.enums import RunMode
|
||||
|
||||
@@ -11,8 +11,8 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup_utils_configuration(
|
||||
args: Dict[str, Any], method: RunMode, *, set_dry: bool = True
|
||||
) -> Dict[str, Any]:
|
||||
args: dict[str, Any], method: RunMode, *, set_dry: bool = True
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Prepare the configuration for utils subcommands
|
||||
:param args: Cli args from Arguments()
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import logging
|
||||
from collections import Counter
|
||||
from copy import deepcopy
|
||||
from typing import Any, Dict
|
||||
from typing import Any
|
||||
|
||||
from jsonschema import Draft4Validator, validators
|
||||
from jsonschema.exceptions import ValidationError, best_match
|
||||
@@ -43,7 +43,7 @@ def _extend_validator(validator_class):
|
||||
FreqtradeValidator = _extend_validator(Draft4Validator)
|
||||
|
||||
|
||||
def validate_config_schema(conf: Dict[str, Any], preliminary: bool = False) -> Dict[str, Any]:
|
||||
def validate_config_schema(conf: dict[str, Any], preliminary: bool = False) -> dict[str, Any]:
|
||||
"""
|
||||
Validate the configuration follow the Config Schema
|
||||
:param conf: Config in JSON format
|
||||
@@ -69,7 +69,7 @@ def validate_config_schema(conf: Dict[str, Any], preliminary: bool = False) -> D
|
||||
raise ValidationError(best_match(Draft4Validator(conf_schema).iter_errors(conf)).message)
|
||||
|
||||
|
||||
def validate_config_consistency(conf: Dict[str, Any], *, preliminary: bool = False) -> None:
|
||||
def validate_config_consistency(conf: dict[str, Any], *, preliminary: bool = False) -> None:
|
||||
"""
|
||||
Validate the configuration consistency.
|
||||
Should be ran after loading both configuration and strategy,
|
||||
@@ -97,7 +97,7 @@ def validate_config_consistency(conf: Dict[str, Any], *, preliminary: bool = Fal
|
||||
validate_config_schema(conf, preliminary=preliminary)
|
||||
|
||||
|
||||
def _validate_unlimited_amount(conf: Dict[str, Any]) -> None:
|
||||
def _validate_unlimited_amount(conf: dict[str, Any]) -> None:
|
||||
"""
|
||||
If edge is disabled, either max_open_trades or stake_amount need to be set.
|
||||
:raise: ConfigurationError if config validation failed
|
||||
@@ -110,7 +110,7 @@ def _validate_unlimited_amount(conf: Dict[str, Any]) -> None:
|
||||
raise ConfigurationError("`max_open_trades` and `stake_amount` cannot both be unlimited.")
|
||||
|
||||
|
||||
def _validate_price_config(conf: Dict[str, Any]) -> None:
|
||||
def _validate_price_config(conf: dict[str, Any]) -> None:
|
||||
"""
|
||||
When using market orders, price sides must be using the "other" side of the price
|
||||
"""
|
||||
@@ -126,7 +126,7 @@ def _validate_price_config(conf: Dict[str, Any]) -> None:
|
||||
raise ConfigurationError('Market exit orders require exit_pricing.price_side = "other".')
|
||||
|
||||
|
||||
def _validate_trailing_stoploss(conf: Dict[str, Any]) -> None:
|
||||
def _validate_trailing_stoploss(conf: dict[str, Any]) -> None:
|
||||
if conf.get("stoploss") == 0.0:
|
||||
raise ConfigurationError(
|
||||
"The config stoploss needs to be different from 0 to avoid problems with sell orders."
|
||||
@@ -159,7 +159,7 @@ def _validate_trailing_stoploss(conf: Dict[str, Any]) -> None:
|
||||
)
|
||||
|
||||
|
||||
def _validate_edge(conf: Dict[str, Any]) -> None:
|
||||
def _validate_edge(conf: dict[str, Any]) -> None:
|
||||
"""
|
||||
Edge and Dynamic whitelist should not both be enabled, since edge overrides dynamic whitelists.
|
||||
"""
|
||||
@@ -173,7 +173,7 @@ def _validate_edge(conf: Dict[str, Any]) -> None:
|
||||
)
|
||||
|
||||
|
||||
def _validate_whitelist(conf: Dict[str, Any]) -> None:
|
||||
def _validate_whitelist(conf: dict[str, Any]) -> None:
|
||||
"""
|
||||
Dynamic whitelist does not require pair_whitelist to be set - however StaticWhitelist does.
|
||||
"""
|
||||
@@ -194,7 +194,7 @@ def _validate_whitelist(conf: Dict[str, Any]) -> None:
|
||||
raise ConfigurationError("StaticPairList requires pair_whitelist to be set.")
|
||||
|
||||
|
||||
def _validate_ask_orderbook(conf: Dict[str, Any]) -> None:
|
||||
def _validate_ask_orderbook(conf: dict[str, Any]) -> None:
|
||||
ask_strategy = conf.get("exit_pricing", {})
|
||||
ob_min = ask_strategy.get("order_book_min")
|
||||
ob_max = ask_strategy.get("order_book_max")
|
||||
@@ -214,7 +214,7 @@ def _validate_ask_orderbook(conf: Dict[str, Any]) -> None:
|
||||
)
|
||||
|
||||
|
||||
def validate_migrated_strategy_settings(conf: Dict[str, Any]) -> None:
|
||||
def validate_migrated_strategy_settings(conf: dict[str, Any]) -> None:
|
||||
_validate_time_in_force(conf)
|
||||
_validate_order_types(conf)
|
||||
_validate_unfilledtimeout(conf)
|
||||
@@ -222,7 +222,7 @@ def validate_migrated_strategy_settings(conf: Dict[str, Any]) -> None:
|
||||
_strategy_settings(conf)
|
||||
|
||||
|
||||
def _validate_time_in_force(conf: Dict[str, Any]) -> None:
|
||||
def _validate_time_in_force(conf: dict[str, Any]) -> None:
|
||||
time_in_force = conf.get("order_time_in_force", {})
|
||||
if "buy" in time_in_force or "sell" in time_in_force:
|
||||
if conf.get("trading_mode", TradingMode.SPOT) != TradingMode.SPOT:
|
||||
@@ -243,7 +243,7 @@ def _validate_time_in_force(conf: Dict[str, Any]) -> None:
|
||||
)
|
||||
|
||||
|
||||
def _validate_order_types(conf: Dict[str, Any]) -> None:
|
||||
def _validate_order_types(conf: dict[str, Any]) -> None:
|
||||
order_types = conf.get("order_types", {})
|
||||
old_order_types = [
|
||||
"buy",
|
||||
@@ -278,7 +278,7 @@ def _validate_order_types(conf: Dict[str, Any]) -> None:
|
||||
process_deprecated_setting(conf, "order_types", o, "order_types", n)
|
||||
|
||||
|
||||
def _validate_unfilledtimeout(conf: Dict[str, Any]) -> None:
|
||||
def _validate_unfilledtimeout(conf: dict[str, Any]) -> None:
|
||||
unfilledtimeout = conf.get("unfilledtimeout", {})
|
||||
if any(x in unfilledtimeout for x in ["buy", "sell"]):
|
||||
if conf.get("trading_mode", TradingMode.SPOT) != TradingMode.SPOT:
|
||||
@@ -297,7 +297,7 @@ def _validate_unfilledtimeout(conf: Dict[str, Any]) -> None:
|
||||
process_deprecated_setting(conf, "unfilledtimeout", o, "unfilledtimeout", n)
|
||||
|
||||
|
||||
def _validate_pricing_rules(conf: Dict[str, Any]) -> None:
|
||||
def _validate_pricing_rules(conf: dict[str, Any]) -> None:
|
||||
if conf.get("ask_strategy") or conf.get("bid_strategy"):
|
||||
if conf.get("trading_mode", TradingMode.SPOT) != TradingMode.SPOT:
|
||||
raise ConfigurationError("Please migrate your pricing settings to use the new wording.")
|
||||
@@ -327,7 +327,7 @@ def _validate_pricing_rules(conf: Dict[str, Any]) -> None:
|
||||
del conf["ask_strategy"]
|
||||
|
||||
|
||||
def _validate_freqai_hyperopt(conf: Dict[str, Any]) -> None:
|
||||
def _validate_freqai_hyperopt(conf: dict[str, Any]) -> None:
|
||||
freqai_enabled = conf.get("freqai", {}).get("enabled", False)
|
||||
analyze_per_epoch = conf.get("analyze_per_epoch", False)
|
||||
if analyze_per_epoch and freqai_enabled:
|
||||
@@ -336,7 +336,7 @@ def _validate_freqai_hyperopt(conf: Dict[str, Any]) -> None:
|
||||
)
|
||||
|
||||
|
||||
def _validate_freqai_include_timeframes(conf: Dict[str, Any], preliminary: bool) -> None:
|
||||
def _validate_freqai_include_timeframes(conf: dict[str, Any], preliminary: bool) -> None:
|
||||
freqai_enabled = conf.get("freqai", {}).get("enabled", False)
|
||||
if freqai_enabled:
|
||||
main_tf = conf.get("timeframe", "5m")
|
||||
@@ -367,7 +367,7 @@ def _validate_freqai_include_timeframes(conf: Dict[str, Any], preliminary: bool)
|
||||
)
|
||||
|
||||
|
||||
def _validate_freqai_backtest(conf: Dict[str, Any]) -> None:
|
||||
def _validate_freqai_backtest(conf: dict[str, Any]) -> None:
|
||||
if conf.get("runmode", RunMode.OTHER) == RunMode.BACKTEST:
|
||||
freqai_enabled = conf.get("freqai", {}).get("enabled", False)
|
||||
timerange = conf.get("timerange")
|
||||
@@ -390,7 +390,7 @@ def _validate_freqai_backtest(conf: Dict[str, Any]) -> None:
|
||||
)
|
||||
|
||||
|
||||
def _validate_consumers(conf: Dict[str, Any]) -> None:
|
||||
def _validate_consumers(conf: dict[str, Any]) -> None:
|
||||
emc_conf = conf.get("external_message_consumer", {})
|
||||
if emc_conf.get("enabled", False):
|
||||
if len(emc_conf.get("producers", [])) < 1:
|
||||
@@ -410,7 +410,7 @@ def _validate_consumers(conf: Dict[str, Any]) -> None:
|
||||
)
|
||||
|
||||
|
||||
def _validate_orderflow(conf: Dict[str, Any]) -> None:
|
||||
def _validate_orderflow(conf: dict[str, Any]) -> None:
|
||||
if conf.get("exchange", {}).get("use_public_trades"):
|
||||
if "orderflow" not in conf:
|
||||
raise ConfigurationError(
|
||||
@@ -418,7 +418,7 @@ def _validate_orderflow(conf: Dict[str, Any]) -> None:
|
||||
)
|
||||
|
||||
|
||||
def _strategy_settings(conf: Dict[str, Any]) -> None:
|
||||
def _strategy_settings(conf: dict[str, Any]) -> None:
|
||||
process_deprecated_setting(conf, None, "use_sell_signal", None, "use_exit_signal")
|
||||
process_deprecated_setting(conf, None, "sell_profit_only", None, "exit_profit_only")
|
||||
process_deprecated_setting(conf, None, "sell_profit_offset", None, "exit_profit_offset")
|
||||
|
||||
@@ -7,7 +7,7 @@ import logging
|
||||
import warnings
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Dict, List, Optional, Tuple
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
from freqtrade import constants
|
||||
from freqtrade.configuration.deprecated_settings import process_temporary_deprecated_settings
|
||||
@@ -37,7 +37,7 @@ class Configuration:
|
||||
Reuse this class for the bot, backtesting, hyperopt and every script that required configuration
|
||||
"""
|
||||
|
||||
def __init__(self, args: Dict[str, Any], runmode: Optional[RunMode] = None) -> None:
|
||||
def __init__(self, args: dict[str, Any], runmode: Optional[RunMode] = None) -> None:
|
||||
self.args = args
|
||||
self.config: Optional[Config] = None
|
||||
self.runmode = runmode
|
||||
@@ -53,7 +53,7 @@ class Configuration:
|
||||
return self.config
|
||||
|
||||
@staticmethod
|
||||
def from_files(files: List[str]) -> Dict[str, Any]:
|
||||
def from_files(files: list[str]) -> dict[str, Any]:
|
||||
"""
|
||||
Iterate through the config files passed in, loading all of them
|
||||
and merging their contents.
|
||||
@@ -68,7 +68,7 @@ class Configuration:
|
||||
c = Configuration({"config": files}, RunMode.OTHER)
|
||||
return c.get_config()
|
||||
|
||||
def load_config(self) -> Dict[str, Any]:
|
||||
def load_config(self) -> dict[str, Any]:
|
||||
"""
|
||||
Extract information for sys.argv and load the bot configuration
|
||||
:return: Configuration dictionary
|
||||
@@ -421,7 +421,7 @@ class Configuration:
|
||||
]
|
||||
self._args_to_config_loop(config, configurations)
|
||||
|
||||
def _args_to_config_loop(self, config, configurations: List[Tuple[str, str]]) -> None:
|
||||
def _args_to_config_loop(self, config, configurations: list[tuple[str, str]]) -> None:
|
||||
for argname, logstring in configurations:
|
||||
self._args_to_config(config, argname=argname, logstring=logstring)
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, Dict
|
||||
from typing import Any
|
||||
|
||||
from freqtrade.constants import ENV_VAR_PREFIX
|
||||
from freqtrade.misc import deep_merge_dicts
|
||||
@@ -24,7 +24,7 @@ def _get_var_typed(val):
|
||||
return val
|
||||
|
||||
|
||||
def _flat_vars_to_nested_dict(env_dict: Dict[str, Any], prefix: str) -> Dict[str, Any]:
|
||||
def _flat_vars_to_nested_dict(env_dict: dict[str, Any], prefix: str) -> dict[str, Any]:
|
||||
"""
|
||||
Environment variables must be prefixed with FREQTRADE.
|
||||
FREQTRADE__{section}__{key}
|
||||
@@ -33,7 +33,7 @@ def _flat_vars_to_nested_dict(env_dict: Dict[str, Any], prefix: str) -> Dict[str
|
||||
:return: Nested dict based on available and relevant variables.
|
||||
"""
|
||||
no_convert = ["CHAT_ID", "PASSWORD"]
|
||||
relevant_vars: Dict[str, Any] = {}
|
||||
relevant_vars: dict[str, Any] = {}
|
||||
|
||||
for env_var, val in sorted(env_dict.items()):
|
||||
if env_var.startswith(prefix):
|
||||
@@ -51,7 +51,7 @@ def _flat_vars_to_nested_dict(env_dict: Dict[str, Any], prefix: str) -> Dict[str
|
||||
return relevant_vars
|
||||
|
||||
|
||||
def enironment_vars_to_dict() -> Dict[str, Any]:
|
||||
def enironment_vars_to_dict() -> dict[str, Any]:
|
||||
"""
|
||||
Read environment variables and return a nested dict for relevant variables
|
||||
Relevant variables must follow the FREQTRADE__{section}__{key} pattern
|
||||
|
||||
@@ -7,7 +7,7 @@ import re
|
||||
import sys
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
from typing import Any, Optional
|
||||
|
||||
import rapidjson
|
||||
|
||||
@@ -42,7 +42,7 @@ def log_config_error_range(path: str, errmsg: str) -> str:
|
||||
return ""
|
||||
|
||||
|
||||
def load_file(path: Path) -> Dict[str, Any]:
|
||||
def load_file(path: Path) -> dict[str, Any]:
|
||||
try:
|
||||
with path.open("r") as file:
|
||||
config = rapidjson.load(file, parse_mode=CONFIG_PARSE_MODE)
|
||||
@@ -51,7 +51,7 @@ def load_file(path: Path) -> Dict[str, Any]:
|
||||
return config
|
||||
|
||||
|
||||
def load_config_file(path: str) -> Dict[str, Any]:
|
||||
def load_config_file(path: str) -> dict[str, Any]:
|
||||
"""
|
||||
Loads a config file from the given path
|
||||
:param path: path as str
|
||||
@@ -78,8 +78,8 @@ def load_config_file(path: str) -> Dict[str, Any]:
|
||||
|
||||
|
||||
def load_from_files(
|
||||
files: List[str], base_path: Optional[Path] = None, level: int = 0
|
||||
) -> Dict[str, Any]:
|
||||
files: list[str], base_path: Optional[Path] = None, level: int = 0
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Recursively load configuration files if specified.
|
||||
Sub-files are assumed to be relative to the initial config.
|
||||
|
||||
Reference in New Issue
Block a user