zwitschi
c8e3daeb57
CI / lint-test-build (push) Failing after 12s
- Consolidated multiline string formatting into single-line for SQL queries in multiple files. - Adjusted argument formatting in function calls for better alignment and readability. - Removed unnecessary line breaks and improved spacing in various sections of the codebase. - Updated test cases to maintain consistency in formatting and improve clarity.
181 lines
8.4 KiB
Python
181 lines
8.4 KiB
Python
from __future__ import annotations
|
|
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
from pydantic import Field, field_validator, model_validator
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
model_config = SettingsConfigDict(
|
|
env_file=".env",
|
|
env_file_encoding="utf-8",
|
|
extra="ignore",
|
|
env_ignore_empty=True,
|
|
)
|
|
|
|
app_env: str = Field(default="dev", alias="APP_ENV")
|
|
app_host: str = Field(default="0.0.0.0", alias="APP_HOST")
|
|
app_port: int = Field(default=9090, alias="APP_PORT")
|
|
|
|
log_level: str = Field(default="INFO", alias="LOG_LEVEL")
|
|
log_json: bool = Field(default=True, alias="LOG_JSON")
|
|
|
|
dashboard_auth_username: str | None = Field(
|
|
default=None,
|
|
alias="DASHBOARD_AUTH_USERNAME",
|
|
)
|
|
dashboard_auth_password: str | None = Field(
|
|
default=None,
|
|
alias="DASHBOARD_AUTH_PASSWORD",
|
|
)
|
|
|
|
alerts_enabled: bool = Field(default=True, alias="ALERTS_ENABLED")
|
|
alert_min_severity: str = Field(default="warning", alias="ALERT_MIN_SEVERITY")
|
|
alert_dedup_seconds: float = Field(default=30.0, alias="ALERT_DEDUP_SECONDS")
|
|
alert_on_trade_events: bool = Field(default=True, alias="ALERT_ON_TRADE_EVENTS")
|
|
alert_on_error_events: bool = Field(default=True, alias="ALERT_ON_ERROR_EVENTS")
|
|
alert_on_threshold_events: bool = Field(default=True, alias="ALERT_ON_THRESHOLD_EVENTS")
|
|
alert_on_system_events: bool = Field(default=True, alias="ALERT_ON_SYSTEM_EVENTS")
|
|
|
|
telegram_alerts_enabled: bool = Field(default=False, alias="TELEGRAM_ALERTS_ENABLED")
|
|
telegram_bot_token: str | None = Field(default=None, alias="TELEGRAM_BOT_TOKEN")
|
|
telegram_chat_id: str | None = Field(default=None, alias="TELEGRAM_CHAT_ID")
|
|
|
|
discord_alerts_enabled: bool = Field(default=False, alias="DISCORD_ALERTS_ENABLED")
|
|
discord_webhook_url: str | None = Field(default=None, alias="DISCORD_WEBHOOK_URL")
|
|
|
|
email_alerts_enabled: bool = Field(default=False, alias="EMAIL_ALERTS_ENABLED")
|
|
email_smtp_host: str | None = Field(default=None, alias="EMAIL_SMTP_HOST")
|
|
email_smtp_port: int = Field(default=587, alias="EMAIL_SMTP_PORT")
|
|
email_smtp_username: str | None = Field(default=None, alias="EMAIL_SMTP_USERNAME")
|
|
email_smtp_password: str | None = Field(default=None, alias="EMAIL_SMTP_PASSWORD")
|
|
email_alert_from: str | None = Field(default=None, alias="EMAIL_ALERT_FROM")
|
|
email_alert_to: str | None = Field(default=None, alias="EMAIL_ALERT_TO")
|
|
email_smtp_use_tls: bool = Field(default=True, alias="EMAIL_SMTP_USE_TLS")
|
|
|
|
duckdb_path: Path = Field(default=Path("./data/arbitrade.duckdb"), alias="DUCKDB_PATH")
|
|
|
|
kraken_rest_url: str = Field(default="https://api.kraken.com", alias="KRAKEN_REST_URL")
|
|
kraken_ws_url: str = Field(default="wss://ws.kraken.com/v2", alias="KRAKEN_WS_URL")
|
|
kraken_private_rate_limit_seconds: float = Field(
|
|
default=1.0, alias="KRAKEN_PRIVATE_RATE_LIMIT_SECONDS"
|
|
)
|
|
kraken_http_timeout_seconds: float = Field(default=10.0, alias="KRAKEN_HTTP_TIMEOUT_SECONDS")
|
|
kraken_retry_attempts: int = Field(default=3, alias="KRAKEN_RETRY_ATTEMPTS")
|
|
kraken_retry_base_delay_seconds: float = Field(
|
|
default=0.25, alias="KRAKEN_RETRY_BASE_DELAY_SECONDS"
|
|
)
|
|
kraken_api_key: str | None = Field(default=None, alias="KRAKEN_API_KEY")
|
|
kraken_api_secret: str | None = Field(default=None, alias="KRAKEN_API_SECRET")
|
|
kraken_api_key_permissions: str = Field(
|
|
default="query,trade",
|
|
alias="KRAKEN_API_KEY_PERMISSIONS",
|
|
)
|
|
ws_heartbeat_timeout_seconds: float = Field(default=20.0, alias="WS_HEARTBEAT_TIMEOUT_SECONDS")
|
|
ws_max_staleness_seconds: float = Field(default=5.0, alias="WS_MAX_STALENESS_SECONDS")
|
|
strategy_enable_stat_arb_experiment: bool = Field(
|
|
default=False,
|
|
alias="STRATEGY_ENABLE_STAT_ARB_EXPERIMENT",
|
|
)
|
|
strategy_stat_arb_lookback_window: int = Field(
|
|
default=120,
|
|
alias="STRATEGY_STAT_ARB_LOOKBACK_WINDOW",
|
|
)
|
|
strategy_stat_arb_entry_zscore: float = Field(
|
|
default=2.0,
|
|
alias="STRATEGY_STAT_ARB_ENTRY_ZSCORE",
|
|
)
|
|
strategy_stat_arb_exit_zscore: float = Field(
|
|
default=0.5,
|
|
alias="STRATEGY_STAT_ARB_EXIT_ZSCORE",
|
|
)
|
|
strategy_stat_arb_max_holding_seconds: float = Field(
|
|
default=900.0,
|
|
alias="STRATEGY_STAT_ARB_MAX_HOLDING_SECONDS",
|
|
)
|
|
paper_trading_mode: bool = Field(default=True, alias="PAPER_TRADING_MODE")
|
|
trade_capital_usd: float = Field(default=100.0, alias="TRADE_CAPITAL_USD")
|
|
max_trade_capital_usd: float = Field(default=100.0, alias="MAX_TRADE_CAPITAL_USD")
|
|
max_concurrent_trades: int | None = Field(default=None, alias="MAX_CONCURRENT_TRADES")
|
|
max_exposure_per_asset_usd: float | None = Field(
|
|
default=None,
|
|
alias="MAX_EXPOSURE_PER_ASSET_USD",
|
|
)
|
|
quote_balance_asset: str = Field(default="USD", alias="QUOTE_BALANCE_ASSET")
|
|
min_order_size_usd: float | None = Field(default=None, alias="MIN_ORDER_SIZE_USD")
|
|
kill_switch_active: bool = Field(default=False, alias="KILL_SWITCH_ACTIVE")
|
|
daily_loss_limit_usd: float | None = Field(default=None, alias="DAILY_LOSS_LIMIT_USD")
|
|
cumulative_loss_limit_usd: float | None = Field(default=None, alias="CUMULATIVE_LOSS_LIMIT_USD")
|
|
max_source_latency_ms: float | None = Field(default=None, alias="MAX_SOURCE_LATENCY_MS")
|
|
max_apply_latency_ms: float | None = Field(default=None, alias="MAX_APPLY_LATENCY_MS")
|
|
max_consecutive_failures: int | None = Field(default=None, alias="MAX_CONSECUTIVE_FAILURES")
|
|
|
|
fernet_key: str | None = Field(default=None, alias="FERNET_KEY")
|
|
|
|
@field_validator("app_env")
|
|
@classmethod
|
|
def _validate_app_env(cls, value: str) -> str:
|
|
normalized = value.strip().lower()
|
|
if normalized not in {"dev", "test", "prod"}:
|
|
raise ValueError("APP_ENV must be one of: dev, test, prod")
|
|
return normalized
|
|
|
|
@field_validator("log_level")
|
|
@classmethod
|
|
def _validate_log_level(cls, value: str) -> str:
|
|
normalized = value.strip().upper()
|
|
if normalized not in {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}:
|
|
raise ValueError("LOG_LEVEL must be one of: DEBUG, INFO, WARNING, ERROR, CRITICAL")
|
|
return normalized
|
|
|
|
@field_validator("alert_min_severity")
|
|
@classmethod
|
|
def _validate_alert_severity(cls, value: str) -> str:
|
|
normalized = value.strip().lower()
|
|
if normalized not in {"info", "warning", "error", "critical"}:
|
|
raise ValueError("ALERT_MIN_SEVERITY must be one of: info, warning, error, critical")
|
|
return normalized
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_security_constraints(self) -> Settings:
|
|
if bool(self.dashboard_auth_username) ^ bool(self.dashboard_auth_password):
|
|
raise ValueError("dashboard auth requires both username and password")
|
|
|
|
if bool(self.kraken_api_key) ^ bool(self.kraken_api_secret):
|
|
raise ValueError("Kraken API auth requires both API key and secret")
|
|
|
|
permissions = {
|
|
token.strip().lower()
|
|
for token in self.kraken_api_key_permissions.split(",")
|
|
if token.strip()
|
|
}
|
|
if permissions and ("query" not in permissions or "trade" not in permissions):
|
|
raise ValueError("KRAKEN_API_KEY_PERMISSIONS must include query and trade")
|
|
if "withdraw" in permissions or "withdrawals" in permissions:
|
|
raise ValueError("KRAKEN_API_KEY_PERMISSIONS must not include withdrawal scope")
|
|
|
|
if self.alert_dedup_seconds < 0.0:
|
|
raise ValueError("ALERT_DEDUP_SECONDS must be >= 0")
|
|
|
|
if self.strategy_stat_arb_lookback_window < 2:
|
|
raise ValueError("STRATEGY_STAT_ARB_LOOKBACK_WINDOW must be >= 2")
|
|
if self.strategy_stat_arb_entry_zscore <= 0.0:
|
|
raise ValueError("STRATEGY_STAT_ARB_ENTRY_ZSCORE must be > 0")
|
|
if self.strategy_stat_arb_exit_zscore < 0.0:
|
|
raise ValueError("STRATEGY_STAT_ARB_EXIT_ZSCORE must be >= 0")
|
|
if self.strategy_stat_arb_entry_zscore <= self.strategy_stat_arb_exit_zscore:
|
|
raise ValueError(
|
|
"STRATEGY_STAT_ARB_ENTRY_ZSCORE must be greater than STRATEGY_STAT_ARB_EXIT_ZSCORE"
|
|
)
|
|
if self.strategy_stat_arb_max_holding_seconds <= 0.0:
|
|
raise ValueError("STRATEGY_STAT_ARB_MAX_HOLDING_SECONDS must be > 0")
|
|
|
|
return self
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_settings() -> Settings:
|
|
return Settings()
|