Strategy service#

ObjectService

Object cache service to persist a picklable object in s3 for the purpose of sharing between users, as well as saving objects into relatively durable storage.

PersistedStrategy

Storage class for a strategy together with some of its meta data for the purpose of persistence into a cache.

StrategyKeyHelper

Helper class to generate strategy key and to parse key string to its components

StrategyService

Strategy cache service to persist strategy in a multi-level cache for the purpose of sharing between users, as well as saving strategy into relatively durable storage.

StrategyServiceInterface

Abstract class for Strategy Service

StrategyServiceNoop

No-Op Strategy service implements all of StrategyServiceInterface, however, the methods simple return None(s).

StrategyServiceProxy

Proxy class to wrap StrategyService to provide lazy initialisation features as well as a NOOP strategy service when service is sets to inactive.

StrategyStream

Decorator to run a strategy live.
sigtech.framework.services.strategy_service.service.create_strategy_and_cache(raw_data: Optional[dict[str, Any]] = None, use_cache: bool = True, strategy_cls: Optional[collections.abc.Callable[[...], Any]] = None, strategy_sector: Optional[str] = None) → Union[sigtech.framework.strategies.strategy.Strategy, sigtech.framework.internal.strategies.base.StrategyInstrument]

Wraps a strategy class’ constructor using strategy service’s decorator and calls the wrapped constructor with the given raw data It looks up the strategy service by assuming the strategy name as ticker + ‘ STRATEGY’

Parameters

  • use_cache – if True attempt to retrieve from strategy service’s cache, otherwise simply create the object

  • raw_data – data dict for the strategy constructor

  • strategy_cls – constructor of Strategy class or a callable which takes raw data dict to create Strateg object

  • strategy_sector – overriding of STRATEGY suffix in strategy_name.

Returns

Strategy object

sigtech.framework.services.strategy_service.service.strategy_service_cache(strategy_name: str)

Decorator to wrap a strategy generating function. The decorator attempts to retrieve the strategy from the strategy service.

If found, the strategy object is simply returned. If not found, the wrapped function will be called, the returned strategy object will be cached via the strategy service when build() or history() is called by the user application.

# Example Usage
@strategy_service_cache('Strategy Name XYZ')
def get_strategy():
    return ReinvestmentStrategy(....)
Parameters

strategy_name – strategy name for looking up at strategy service.

sigtech.framework.services.strategy_service.service.object_service_cache(name: str, add_env_date_to_key: bool = True, add_data_date_to_key: bool = False, expires: Optional[str] = '30d')

Decorator to wrap an object generating function. The decorator attempts to retrieve the object from the object service.

If found, the object is simply returned. If not found, the wrapped function will be called, the returned object will be cached via the object service.

# Example Usage
@object_service_cache('Some Name')
def get_frame():
    return pd.Dataframe()
Parameters

  • name – name of the object

  • add_env_date_to_key – append the current environment date to the cache key

  • add_data_date_to_key – append the current data date to the cache key

  • expires – allow the data to expire after this time window. None means don’t expire.

sigtech.framework.strategies.runner.metaflow.metadata.setup_metadata(*args, workspace_id=None, organisation_bucket=None, datastore_root=None, provider='api')

Setup the metadata.

Parameters

  • workspace_id – Workspace identifier (optional).

  • organisation_bucket – Organisation bucket (optional).

  • datastore_root – Datastore root (optional).

  • provider – Provider (optional).

sigtech.framework.strategies.runner.metaflow.decorators.strategy_job.get_object_run(collect: list[collections.abc.Callable], outputs: list[Type[sigtech.framework.strategies.runner.output_types.Output]], config_params: Optional[dict] = None, run_kwargs: Optional[dict] = None, pre_run_wait: Optional[collections.abc.Callable] = None) → sigtech.framework.internal.strategies.runner.object_run.ObjectRun

Create an ObjectRun instance.