Skip to content

trading_agent.doctor.checks.unified_broker_interface_check

Checks that tradingmachine's settings file exists and UBI answers.

Any HTTP answer, even an error status, means UBI is running. Only a refused or timed-out connection counts as a failure.

Typical usage example:

result = UnifiedBrokerInterfaceCheck(Path('/home/pramod/Projects/tradingmachine/.env')).run()

UnifiedBrokerInterfaceCheck 

UnifiedBrokerInterfaceCheck(environment_file)

Bases: DoctorCheck

Checks the route from the agent to UBI through tradingmachine.

Creates the check.

Parameters:

Name Type Description Default
environment_file Path

The .env file tradingmachine reads its UBI address from.

 required
Source code in src/trading_agent/doctor/checks/unified_broker_interface_check.py 
27
28
29
30
31
32
33
def __init__(self, environment_file: Path):
    """Creates the check.

    Args:
        environment_file (Path): The .env file tradingmachine reads its UBI address from.
    """
    self._environment_file = environment_file

run 

run()

Reads UBI's address from tradingmachine and sends it one request.

Returns:

Type Description
CheckResult

OK when UBI answers, failed when the settings file is missing or UBI cannot be reached.
Source code in src/trading_agent/doctor/checks/unified_broker_interface_check.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def run(self) -> doctor_check.CheckResult:
    """Reads UBI's address from tradingmachine and sends it one request.

    Returns:
        doctor_check.CheckResult: OK when UBI answers, failed when the settings file is missing or UBI cannot be reached.
    """
    if not self._environment_file.is_file():
        return doctor_check.CheckResult(
            doctor_check.CheckStatus.FAILED,
            f'tradingmachine settings file not found: {self._environment_file}',
        )
    settings = tradingmachine_configuration.Configuration(
        environment_file=str(self._environment_file),
    )
    base_url = settings.ubi_base_url
    try:
        response = httpx.get(base_url, timeout=_TIMEOUT_SECONDS)
    except httpx.HTTPError as error:
        return doctor_check.CheckResult(
            doctor_check.CheckStatus.FAILED,
            f'UBI at {base_url} did not answer: {error!r}',
        )
    return doctor_check.CheckResult(
        doctor_check.CheckStatus.OK,
        f'UBI at {base_url} answered (HTTP {response.status_code}).',
    )