Source code for tradeexecutor.cli.commands.webapi

"""Web API command

Example::

    poetry run trade-executor webapi --strategy-file=strategies/enzyme-polygon-matic-usdc.py
"""

import datetime
import faulthandler
import logging
import os
import time
from decimal import Decimal 
from pathlib import Path
from queue import Queue
from typing import Optional

import typer
try:
    import waitress
    from ...webhook.server import create_pyramid_app
except ImportError:
    waitress = None

from . import shared_options
from .app import app
from ..bootstrap import prepare_executor_id, prepare_cache, create_metadata, create_state_store
from ..log import setup_logging, setup_discord_logging, setup_logstash_logging, setup_file_logging
from ..version_info import VersionInfo
from ...state.state import State
from ...strategy.execution_model import AssetManagementMode
from ...strategy.run_state import RunState
from ...strategy.strategy_module import read_strategy_module, StrategyModuleInformation
from ...statistics.in_memory_statistics import refresh_run_state
from ...strategy.execution_context import ExecutionContext, ExecutionMode
from ...strategy.cycle import CycleDuration


logger = logging.getLogger(__name__)


[docs]@app.command() def webapi( # Strategy assets id: str = shared_options.id, name: Optional[str] = shared_options.name, short_description: Optional[str] = typer.Option(None, envvar="SHORT_DESCRIPTION", help="Short description for metadata"), long_description: Optional[str] = typer.Option(None, envvar="LONG_DESCRIPTION", help="Long description for metadata"), badges: Optional[str] = typer.Option(None, envvar="BADGES", help="Comma separated list of badges to be displayed on the strategy tile"), icon_url: Optional[str] = typer.Option(None, envvar="ICON_URL", help="Strategy icon for web rendering and Discord avatar"), strategy_file: Path = shared_options.strategy_file, # Webhook server options http_port: int = typer.Option(3456, envvar="HTTP_PORT", help="Which HTTP port to listen. The default is 3456, the default port of Pyramid web server."), http_host: str = typer.Option("0.0.0.0", envvar="HTTP_HOST", help="The IP address to bind for the web server. By default listen to all IP addresses available in the run-time environment."), http_username: str = typer.Option(None, envvar="HTTP_USERNAME", help="Username for HTTP Basic Auth protection of webhooks"), http_password: str = typer.Option(None, envvar="HTTP_PASSWORD", help="Password for HTTP Basic Auth protection of webhooks"), # Logging file_log_level: Optional[str] = typer.Option("info", envvar="FILE_LOG_LEVEL", help="Log file log level. The default log file is logs/id.log."), # Logging log_level: str = shared_options.log_level, # Various file configurations state_file: Optional[Path] = shared_options.state_file, cache_path: Optional[Path] = shared_options.cache_path, ): """Launch Trade Executor instance.""" global logger started_at = datetime.datetime.utcnow() # Guess id from the strategy file id = prepare_executor_id(id, strategy_file) # We always need a name-*- if not name: if strategy_file: name = os.path.basename(strategy_file) else: name = "Unnamed backtest" if not log_level: log_level = logging.INFO # Make sure unit tests run logs do not get polluted # Don't touch any log levels, but # make sure we have logger.trading() available when # log_level is "disabled" logger = setup_logging(log_level, in_memory_buffer=True) setup_file_logging( f"logs/{id}.log", file_log_level, http_logging=True, ) if not state_file: state_file = f"state/{id}.json" cache_path = prepare_cache(id, cache_path, False) mod = read_strategy_module(strategy_file) if state_file: store = create_state_store(Path(state_file)) fees = dict( management_fee=mod.management_fee, trading_strategy_protocol_fee=mod.trading_strategy_protocol_fee, strategy_developer_fee=mod.strategy_developer_fee, ) metadata = create_metadata( name, short_description, long_description, icon_url, AssetManagementMode.dummy, chain_id=mod.get_default_chain_id(), vault=None, fees=fees, ) # Start the queue that relays info from the web server to the strategy executor command_queue = Queue() run_state = RunState() run_state.version = VersionInfo.read_docker_version() run_state.executor_id = id # Set up read-only state sync if not store.is_pristine(): run_state.read_only_state_copy = store.load() refresh_run_state( run_state, store.load(), ExecutionContext(mode=ExecutionMode.unit_testing), cycle_duration=mod.parameters.cycle_duration, ) app = create_pyramid_app( http_username, http_password, command_queue, store, metadata, run_state, ) waitress.serve(app, host=http_host, port=http_port)