Skip to content

trading_agent.tools.set_price_alert_tool

The set_price_alert tool: asks the market watcher to wake the desk when a price is reached.

Typical usage example:

tool = SetPriceAlertTool(reader, store)

SetPriceAlertArguments

Bases: InstrumentArguments

The arguments of set_price_alert.

Attributes:

Name Type Description
condition Literal['above', 'below']

"above" to fire when the price rises to or past it, "below" when it falls to or past it.
price float

The price, in rupees.
note str

What you want to do when it fires, so your future self knows.

SetPriceAlertTool 

SetPriceAlertTool(reader, store)

Bases: Tool

Adds a price alert.

Creates the tool.

Parameters:

Name Type Description Default
reader BrokerReader

The shared UBI reader.

 required
store PriceAlertStore

The desk's alerts.

 required
Source code in src/trading_agent/tools/set_price_alert_tool.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    reader: broker_reader_module.BrokerReader,
    store: price_alert_store_module.PriceAlertStore,
):
    """Creates the tool.

    Args:
        reader (broker_reader_module.BrokerReader): The shared UBI reader.
        store (price_alert_store_module.PriceAlertStore): The desk's alerts.
    """
    self._reader = reader
    self._store = store

run async 

run(arguments)

Checks the instrument and the current price, then stores the alert.

Parameters:

Name Type Description Default
arguments SetPriceAlertArguments

The instrument, side, price and note.

 required

Returns:

Type Description
ToolResult

The alert and every active alert, or why it was refused.
Source code in src/trading_agent/tools/set_price_alert_tool.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def run(self, arguments: SetPriceAlertArguments) -> tool.ToolResult:
    """Checks the instrument and the current price, then stores the alert.

    Args:
        arguments (SetPriceAlertArguments): The instrument, side, price and note.

    Returns:
        tool.ToolResult: The alert and every active alert, or why it was refused.
    """
    try:
        instrument = await self._reader.call(
            functools.partial(self._reader.tradeable_instrument, arguments)
        )
        last_price = await self._reader.call(
            functools.partial(self._last_price, instrument)
        )
    except (
        ubi_exceptions.UnifiedBrokerInterfaceError,
        instrument_exceptions.InstrumentError,
    ) as error:
        return tool.ToolResult.error(f'{type(error).__name__}: {error}')
    alert = price_alert_store_module.PriceAlert.new(
        instrument_id=instrument.instrument_id,
        exchange=instrument.exchange,
        segment=instrument.segment,
        symbol=instrument.symbol or instrument.underlying_symbol or '',
        condition=arguments.condition,
        price=arguments.price,
        note=arguments.note,
    )
    if last_price is not None and alert.is_crossed(last_price):
        return tool.ToolResult.error(
            f'{alert.symbol} is already at {last_price:,.2f}, {arguments.condition} {arguments.price:,.2f}, so the alert would fire at once.'
        )
    self._store.add(alert)
    lines = [
        f'Alert {alert.alert_id} set: {alert.describe()}. The last price is {last_price}.',
        'Your active alerts:',
    ]
    for active in self._store.alerts():
        lines.append(f'- {active.alert_id}: {active.describe()}')
    return tool.ToolResult.success('\n'.join(lines))