credmark.cmf.model.IncrementalModel

class IncrementalModel(context)

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

convert_dict_to_dto	A model can call this method to convert a dict of data in a known format into a DTO instance.
describe	Decorator for credmark.cmf.model.IncrementalModel subclasses to describe the model.
init	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.

Parameters

• data (dict) –

• dto_class (Type[Union[BaseModel, IntDTO, StrDTO, FloatDTO]]) –

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)

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.

• display_name (Optional[str]) – Name for the model

• 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 to EmptyInput 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 configured ModelDataErrorDesc instance (or a list of instances) describing the errors. Defaults to None.

init()

Subclasses may override this method to do any model instance initiation.