Portfolio

class zvt.contract.schema.Portfolio

Bases: zvt.contract.schema.TradableEntity

composition of tradable entities

__init__()
classmethod get_stocks(code=None, codes=None, ids=None, timestamp=Timestamp('2024-03-14 14:55:20.130905'), provider=None)

the publishing policy of portfolio positions is different for different types, overwrite this function for get the holding stocks in specific date

Parameters
  • code – portfolio(etf/block/index…) code

  • codes – portfolio(etf/block/index…) codes

  • ids – portfolio(etf/block/index…) ids

  • timestamp – the date of the holding stocks

  • provider – the data provider

Returns

classmethod could_short()

whether could be shorted

Returns

classmethod get_interval_timestamps(start_date, end_date, level: zvt.contract.IntervalLevel)

generate the timestamps for the level

Parameters
  • start_date

  • end_date

  • level

classmethod get_providers() List[str]

providers of the schema defined by cls

Returns

providers

classmethod get_storages(provider: Optional[str] = None)

get the storages info

Parameters

provider – provider

Returns

storages

classmethod get_trading_dates(start_date=None, end_date=None)

overwrite it to get the trading dates of the entity

Parameters
  • start_date

  • end_date

Returns

list of dates

classmethod get_trading_intervals()

overwrite it to get the trading intervals of the entity

Returns

list of time intervals, in format [(start,end)]

classmethod get_trading_t()

0 means t+0 1 means t+1

Returns

classmethod is_finished_kdata_timestamp(timestamp: pandas._libs.tslibs.timestamps.Timestamp, level: zvt.contract.IntervalLevel)
Parameters
  • timestamp (pd.Timestamp) – the timestamp could be recorded in kdata of the level

  • level (zvt.domain.common.IntervalLevel) –

Returns

Return type

bool

classmethod query_data(provider_index: int = 0, ids: Optional[List[str]] = None, entity_ids: Optional[List[str]] = None, entity_id: Optional[str] = None, codes: Optional[List[str]] = None, code: Optional[str] = None, level: Optional[Union[zvt.contract.IntervalLevel, str]] = None, provider: Optional[str] = None, columns: Optional[List] = None, col_label: Optional[dict] = None, return_type: str = 'df', start_timestamp: Optional[Union[str, pandas._libs.tslibs.timestamps.Timestamp]] = None, end_timestamp: Optional[Union[str, pandas._libs.tslibs.timestamps.Timestamp]] = None, filters: Optional[List] = None, session: Optional[sqlalchemy.orm.session.Session] = None, order=None, limit: Optional[int] = None, index: Optional[Union[str, list]] = None, drop_index_col=False, time_field: str = 'timestamp')

query data by the arguments

Parameters
  • provider_index

  • data_schema

  • ids

  • entity_ids

  • entity_id

  • codes

  • code

  • level

  • provider

  • columns

  • col_label – dict with key(column), value(label)

  • return_type – df, domain or dict. default is df

  • start_timestamp

  • end_timestamp

  • filters

  • session

  • order

  • limit

  • index – index field name, str for single index, str list for multiple index

  • drop_index_col – whether drop the col if it’s in index, default False

  • time_field

Returns

results basing on return_type.

classmethod record_data(provider_index: int = 0, provider: Optional[str] = None, force_update=None, sleeping_time=None, exchanges=None, entity_id=None, entity_ids=None, code=None, codes=None, real_time=None, fix_duplicate_way=None, start_timestamp=None, end_timestamp=None, one_day_trading_minutes=None, **kwargs)

record data by the arguments

Parameters
  • provider_index

  • provider

  • force_update

  • sleeping_time

  • exchanges

  • entity_ids

  • code

  • codes

  • real_time

  • fix_duplicate_way

  • start_timestamp

  • end_timestamp

  • one_day_trading_minutes

  • kwargs

Returns

classmethod register_provider(provider)

register the provider to the schema defined by cls

Parameters

provider

classmethod register_recorder_cls(provider, recorder_cls)

register the recorder for the schema

Parameters
  • provider

  • recorder_cls