credmark.cmf.model.IncrementalModel
- class IncrementalModel(context)[source]
Bases:
BaseModel
Use IncrementalModel for models that only append to output with new blocks. Behind the scenes it uses CachePolicy.INCREMENTAL cache. These models should always return a BlockSeries. Use additional from_block arg in run method to only return BlockSeries for from_block -> context.block.
Methods
A model can call this method to convert a dict of data in a known format into a DTO instance.
Decorator for credmark.cmf.model.IncrementalModel subclasses to describe the model.
Subclasses may override this method to do any model instance initiation.
run
Attributes
- Parameters
context (ModelContext) –
- convert_dict_to_dto(data, dto_class)
A model can call this method to convert a dict of data in a known format into a DTO instance.
- classmethod describe(*, slug, version, display_name=None, description=None, developer=None, category=None, subcategory=None, tags=None, input=<class 'credmark.dto.EmptyInput'>, output=<class 'credmark.cmf.types.series.BlockSeries'>, errors=None)[source]
Decorator for credmark.cmf.model.IncrementalModel subclasses to describe the model.
It shall return a parameterized class of BlockSeries with below requirements: 1. Results with block number less than or equal to the current block number 2. Results with block number greater than or equal to from_block. 3. blockNumber >= 0
BlockSeries object has a method .to_range(from_block, to_block) to filter for [from_block, to_block]
Example usage:
from credmark.cmf.model import IncrementalModel from credmark.cmf.types.series import BlockSeries @IncrementalModel.describe(slug='example.echo-inc', version='1.0', display_name='Echo', description="A test model to echo the message property sent in input.", developer="my_username", category="financial", input=EchoDto, output=BlockSeries[int]) class EchoModel(IncrementalModel): def run(self, input: EchoDto, from_block: BlockNumber) -> BlockSeries[int]: return BlockSeries(series=[ BlockSeriesRow( blockNumber=0, blockTimestamp=0, sampleTimestamp=0, output=1 ), BlockSeriesRow( blockNumber=1, blockTimestamp=1, sampleTimestamp=1, output=2 ), ])
- Parameters
slug (str) – slug (short unique name) for the model. Can contain alpha-numeric and and underscores. Contributor slugs must start with
"contrib."
Once submitted, the model slug cannot be changed.version (str) – version string, ex.
"1.0"
. The version number can be incremented when the model code is updated.description (Optional[str]) – Description of the model. If description is not set, the doc string (
__doc__
) of the model class is used instead.developer (Optional[str]) – Name or nickname of the developer
category (Optional[str]) – Category of the model (ex. “financial”, “protocol” etc.)
subcategory (Optional[str]) – Optional subcategory (ex. “aave”)
tags (Optional[list[str]]) – optional list of string tags describing the model
input (Union[Type[Union[BaseModel, IntDTO, StrDTO, FloatDTO]], Type[dict]]) – Type that model uses as input; a
DTO
subclass or dict. Defaults toEmptyInput
object.output (Optional[Type[BlockSeries]]) – Type that the model run returns; a
BlockSeries
subclass.errors (Optional[Union[List[ModelErrorDesc], ModelErrorDesc]]) – If the model raises
ModelDataError
, set this to a configuredModelDataErrorDesc
instance (or a list of instances) describing the errors. Defaults to None.
- init()
Subclasses may override this method to do any model instance initiation.