trading_agent.scheduling.price_alert_store

Keeps one desk's price alerts: wake me when this instrument trades above or below a price.

Alerts live in desks/<desk>/alerts.json. The market watcher checks them every cycle while the market is open, wakes the desk when one is crossed, and removes it, so each alert fires once.

Typical usage example:

store = PriceAlertStore(Path('workspace/desks/main/alerts.json')) store.add(alert)

ABOVE module-attribute

ABOVE = 'above'

BELOW module-attribute

BELOW = 'below'

PriceAlert dataclass

PriceAlert(alert_id, instrument_id, exchange, segment, symbol, condition, price, note, created_at)

One price alert.

Attributes:

Name	Type	Description
alert_id	str	A short identifier.
instrument_id	str	UBI's id for the instrument.
exchange	str	The instrument's exchange.
segment	str	Its segment.
symbol	str	Its symbol, for messages.
condition	str	"above" or "below".
price	float	The price that fires the alert.
note	str	What the desk wanted to do when it fires.
created_at	str	When it was set, as an ISO 8601 string.

new classmethod

new(**fields)

Creates an alert with a fresh identifier and creation time.

Parameters:

Name	Type	Description	Default
**fields	Any	The other fields.	{}

Returns:

Type	Description
PriceAlert	The alert.

Source code in src/trading_agent/scheduling/price_alert_store.py

@classmethod
def new(cls, **fields: Any) -> Self:
    """Creates an alert with a fresh identifier and creation time.

    Args:
        **fields (Any): The other fields.

    Returns:
        PriceAlert: The alert.
    """
    return cls(
        alert_id=uuid.uuid4().hex[:8],
        created_at=datetime.datetime.now(_INDIA).isoformat(
            timespec='seconds'
        ),
        **fields,
    )

is_crossed

is_crossed(last_price)

Says whether a price fires the alert.

Parameters:

Name	Type	Description	Default
last_price	float	The latest traded price.	required

Returns:

Type	Description
bool	True when the price is at or beyond the alert's price on its side.

Source code in src/trading_agent/scheduling/price_alert_store.py

def is_crossed(self, last_price: float) -> bool:
    """Says whether a price fires the alert.

    Args:
        last_price (float): The latest traded price.

    Returns:
        bool: True when the price is at or beyond the alert's price on its side.
    """
    if self.condition == ABOVE:
        return last_price >= self.price
    return last_price <= self.price

describe

describe()

The alert in one line.

Returns:

Type	Description
str	Such as "INFY above 1,175.00 (sell half)".

Source code in src/trading_agent/scheduling/price_alert_store.py

def describe(self) -> str:
    """The alert in one line.

    Returns:
        str: Such as "INFY above 1,175.00 (sell half)".
    """
    note = f' ({self.note})' if self.note else ''
    return f'{self.symbol} {self.condition} {self.price:,.2f}{note}'

PriceAlertStore

PriceAlertStore(path)

One desk's alerts file.

Describes the file without touching the disk.

Parameters:

Name	Type	Description	Default
path	Path	The desk's alerts.json.	required

Source code in src/trading_agent/scheduling/price_alert_store.py

def __init__(self, path: Path):
    """Describes the file without touching the disk.

    Args:
        path (Path): The desk's alerts.json.
    """
    self._path = path

alerts

alerts()

Reads the alerts; a damaged entry is logged and left out.

Returns:

Type	Description
list[PriceAlert]	The alerts, oldest first.

Source code in src/trading_agent/scheduling/price_alert_store.py

def alerts(self) -> list[PriceAlert]:
    """Reads the alerts; a damaged entry is logged and left out.

    Returns:
        list[PriceAlert]: The alerts, oldest first.
    """
    if not self._path.is_file():
        return []
    try:
        entries = json.loads(self._path.read_text(encoding='utf-8'))
    except OSError, ValueError:
        _LOGGER.exception('Could not read %s', self._path)
        return []
    alerts = []
    for entry in entries:
        try:
            alerts.append(PriceAlert(**entry))
        except TypeError:
            _LOGGER.warning('Skipped a damaged alert in %s', self._path)
    return alerts

add

add(alert)

Adds an alert.

Parameters:

Name	Type	Description	Default
alert	PriceAlert	The alert.	required

Source code in src/trading_agent/scheduling/price_alert_store.py

def add(self, alert: PriceAlert) -> None:
    """Adds an alert.

    Args:
        alert (PriceAlert): The alert.
    """
    alerts = self.alerts()
    alerts.append(alert)
    self._write(alerts)

remove

remove(alert_id)

Removes an alert.

Parameters:

Name	Type	Description	Default
alert_id	str	The alert's identifier.	required

Returns:

Type	Description
PriceAlert | None	The removed alert, or None when there was none with that id.

Source code in src/trading_agent/scheduling/price_alert_store.py

def remove(self, alert_id: str) -> PriceAlert | None:
    """Removes an alert.

    Args:
        alert_id (str): The alert's identifier.

    Returns:
        PriceAlert | None: The removed alert, or None when there was none with that id.
    """
    kept = []
    removed = None
    for alert in self.alerts():
        if alert.alert_id == alert_id:
            removed = alert
        else:
            kept.append(alert)
    if removed is not None:
        self._write(kept)
    return removed