Luna Model
A Luna model is a custom data type which can be used within Template
definitions at runtime for:
- Event payload structures
- Template Parameters
- Template Variables
- general use in Clause logic
It is most simple to think of LunaModel
as a pydantic BaseModel
(because this is what is extended).
A LunaModel
can be nested into others if desired.
Below is an example of a Template which uses Luna Models to manage simple market data, with some light randomization.
import random
from decimal import Decimal
import luna
from luna.types import LunaModel
class Tag(LunaModel):
name: str
dt: datetime
class MktData(LunaModel):
stock: str
price: Decimal
price_time: datetime
tags: list[Tag]
class Currency(LunaModel):
"""
Represents a currency with its name, code, and exchange rate to USD.
Attributes:
- `name` (str): The name of the currency (e.g., "US dollar").
- `code` (str): The currency code (e.g., "USD").
- `usd_exchange_rate` (Decimal): The exchange rate of this currency to USD.
"""
name: str
code: str
usd_exchange_rate: Decimal # how much to get one usd
class Security(LunaModel):
"""
Represents a financial security with its reference, base currency, and price.
Attributes:
- `ref` (str): The reference identifier for the security (e.g., "MSFT").
- `base_currency` (str): The base currency of the security (e.g., "USD").
- `price` (Decimal): The price of the security.
"""
ref: str
base_currency: str
price: Decimal # how much to get one usd
class MarketData(luna.Template):
"""
This template models market data management, including the dynamic handling of currency and security data.
Features include:
- Definition of currency and security data structures using LunaModel.
- Dynamic updating of currency exchange rates and security prices at regular intervals.
- Event-driven methods for setting up initial data for currencies and securities.
"""
broker = luna.Role()
currencies = luna.Variable(type_=list[Currency], value=[])
securities = luna.Variable(type_=list[Security], value=[])
@luna.clause(
on=luna.Event(name="setup_currency", payload={"value": list[Currency]})
)
def create_currencies(self, ctx, event):
"""
Creates or updates the list of currencies based on the provided event payload.
**Parameters:**
- `ctx` (Context): Provides access to the current state, environment, and utilities of the Luna execution context.
- `event` (Event): The event containing the currency data to be set up or updated.
"""
res = [x for x in self.currencies]
for currency in event.payload.value:
idx, _ = next(
((i, c) for (i, c) in enumerate(res) if c.code == currency.code),
(None, None),
)
if idx:
del res[idx]
res += [currency]
self.currencies = res
@luna.clause(
on=luna.Event(name="setup_securities", payload={"value": list[Security]})
)
def create_securities(self, ctx, event):
"""
Creates or updates the list of securities based on the provided event payload.
**Parameters:**
- `ctx` (Context): Provides access to the current state, environment, and utilities of the Luna execution context.
- `event` (Event): The event containing the security data to be set up or updated.
"""
res = [x for x in self.securities]
for security in event.payload.value:
idx, _ = next(
(
(i, c)
for (i, c) in enumerate(self.securities)
if c.ref == security.ref
),
(None, None),
)
if idx:
del res[idx]
res += [security]
self.securities = res
@luna.clause(on=luna.Every("30 minutes"))
def update_currency(self, ctx):
"""
Periodically updates the exchange rates of the currencies.
**Parameters:**
- `ctx` (Context): Provides access to the current state, environment, and utilities of the Luna execution context.
"""
res = []
for currency in self.currencies:
if currency.code.lower() == "usd":
res.append(currency)
continue
r = Decimal(random.randrange(1, 200)) / 10000
r *= random.randint(-1, 1)
ccy_amt = Decimal(currency.usd_exchange_rate) * (1 + r)
res.append(
Currency(
name=currency.name,
code=currency.code,
usd_exchange_rate=ccy_amt,
)
)
self.currencies = res
@luna.clause(on=luna.Every("30 minutes"))
def update_security(self, ctx):
"""
Periodically updates the prices of the securities.
**Parameters:**
- `ctx` (Context): Provides access to the current state, environment, and utilities of the Luna execution context.
"""
res = []
for security in self.securities:
r = Decimal(random.randrange(1, 200)) / 10000
r *= random.choice([-1, 1])
s = Decimal(security.price) * (1 + r)
res.append(
Security(
ref=security.ref,
base_currency=security.base_currency,
price=s,
)
)
self.securities = res