TechnicalFactor

class zvt.factors.technical_factor.TechnicalFactor(entity_schema: typing.Type[zvt.contract.schema.TradableEntity] = <class 'zvt.domain.meta.stock_meta.Stock'>, provider: typing.Optional[str] = None, entity_provider: typing.Optional[str] = None, entity_ids: typing.Optional[typing.List[str]] = None, exchanges: typing.Optional[typing.List[str]] = None, codes: typing.Optional[typing.List[str]] = None, start_timestamp: typing.Optional[typing.Union[str, pandas._libs.tslibs.timestamps.Timestamp]] = None, end_timestamp: typing.Optional[typing.Union[str, pandas._libs.tslibs.timestamps.Timestamp]] = None, columns: typing.Optional[typing.List] = None, filters: typing.Optional[typing.List] = None, order: typing.Optional[object] = None, limit: typing.Optional[int] = None, level: typing.Union[str, zvt.contract.IntervalLevel] = IntervalLevel.LEVEL_1DAY, category_field: str = 'entity_id', time_field: str = 'timestamp', keep_window: typing.Optional[int] = None, keep_all_timestamp: bool = False, fill_method: str = 'ffill', effective_number: typing.Optional[int] = None, transformer: typing.Optional[zvt.contract.factor.Transformer] = None, accumulator: typing.Optional[zvt.contract.factor.Accumulator] = None, need_persist: bool = False, only_compute_factor: bool = False, factor_name: typing.Optional[str] = None, clear_state: bool = False, only_load_factor: bool = False, adjust_type: typing.Optional[typing.Union[zvt.contract.AdjustType, str]] = None)

Bases: zvt.contract.factor.Factor

__init__(entity_schema: typing.Type[zvt.contract.schema.TradableEntity] = <class 'zvt.domain.meta.stock_meta.Stock'>, provider: typing.Optional[str] = None, entity_provider: typing.Optional[str] = None, entity_ids: typing.Optional[typing.List[str]] = None, exchanges: typing.Optional[typing.List[str]] = None, codes: typing.Optional[typing.List[str]] = None, start_timestamp: typing.Optional[typing.Union[str, pandas._libs.tslibs.timestamps.Timestamp]] = None, end_timestamp: typing.Optional[typing.Union[str, pandas._libs.tslibs.timestamps.Timestamp]] = None, columns: typing.Optional[typing.List] = None, filters: typing.Optional[typing.List] = None, order: typing.Optional[object] = None, limit: typing.Optional[int] = None, level: typing.Union[str, zvt.contract.IntervalLevel] = IntervalLevel.LEVEL_1DAY, category_field: str = 'entity_id', time_field: str = 'timestamp', keep_window: typing.Optional[int] = None, keep_all_timestamp: bool = False, fill_method: str = 'ffill', effective_number: typing.Optional[int] = None, transformer: typing.Optional[zvt.contract.factor.Transformer] = None, accumulator: typing.Optional[zvt.contract.factor.Accumulator] = None, need_persist: bool = False, only_compute_factor: bool = False, factor_name: typing.Optional[str] = None, clear_state: bool = False, only_load_factor: bool = False, adjust_type: typing.Optional[typing.Union[zvt.contract.AdjustType, str]] = None) None
Parameters
  • keep_all_timestamp

  • fill_method

  • effective_number

  • transformer

  • accumulator

  • need_persist – whether persist factor

  • only_compute_factor – only compute factor nor result

  • factor_name

  • clear_state

  • only_load_factor – only load factor and compute result

clear_state_data(entity_id=None)

clear state of the entity

Parameters

entity_id – entity id

decode_state(state: str)

decode state

Parameters

state

Returns

encode_state(state: object)

encode state

Parameters

state

Returns

factor_col_map_object_hook() dict

:return:{col:object_hook}

move_on(to_timestamp: Optional[Union[str, pandas._libs.tslibs.timestamps.Timestamp]] = None, timeout: int = 20) object

using continual fetching data in realtime 1)get the data happened before to_timestamp,if not set,get all the data which means to now 2)if computing_window set,the data_df would be cut for saving memory

Parameters
  • to_timestamp

  • timeout (int) –

Returns

Return type

on_data_changed(data: pandas.core.frame.DataFrame)

overwrite it for computing after data added

Parameters

data

on_data_loaded(data: pandas.core.frame.DataFrame)
Parameters

data

on_entity_data_changed(entity, added_data: pandas.core.frame.DataFrame)

overwrite it for computing after entity data added

Parameters
  • entity

  • added_data

state_schema

alias of zvt.contract.zvt_info.FactorState