8. Operator 交易員類

Operator(交易員)類，用於生成Operator對象，qteasy的核心對象。

Operator是一個策略容器，它包含一系列交易策略，保存每一個交易策略所需的歷史數據，並且可以調用所有交易策略，生成交易信號，同時根據保存的規則把所有交易策略生成的信號混合起來，形成一組最終的交易信號。就像一個交易員在實際交易中的行爲一樣。

創建一個Operator對象時，需要給出一組交易策略，並設定好交易員的交易模式和信號模式，交易模式和信號模式都是Operator對象最重要的屬性，他們共同決定了交易員的行爲模式:

class qteasy.Operator(strategies: Optional[Union[str, BaseStrategy, type, list[Union[str, qteasy.strategy.BaseStrategy, type]]]] = None, *, name: Optional[str] = None, signal_type: str = 'pt', op_type: str = 'batch', group_merge_type: str = 'None', run_freq: str = 'd', run_timing: str = 'close')[原始碼]

qteasy 中的核心「交易員」對象，用於承載策略組並在統一時間表上生成交易信號。

Operator 負責管理一個或多個策略組，準備並緩存運行所需的歷史數據，按照運行頻率與時機調用各策略生成信號，並按組內/組間合併規則將信號合成爲最終的交易指令，可用於回測、優化和實盤運行。關於 Operator 與 Strategy / Group 的關係、run_freq / run_timing 單一來源等架構細節，詳見文檔「Operator / Strategy / Group 架構」章節。

add_strategies(strategies: Union[str, list[Union[str, qteasy.strategy.BaseStrategy, type]]], run_freq: str = 'd', run_timing: str = 'close', **kwargs: Any)[原始碼]

添加多個Strategy交易策略到Operator對象中

使用這個方法，不能在添加交易策策略的同時修改交易策略的基本屬性輸入參數strategies可以爲一個列表或者一個逗號分隔字符串列表中的元素可以爲代表內置策略類型的字符串，或者爲一個具體的策略對象字符串和策略對象可以混合給出

參數:

strategies (stg or list of str or list of Strategy) – 交易策略的名稱或者交易策略對象
run_freq (str, optional) – run_freq爲策略的運行頻率，可以爲None，表示不指定運行頻率
run_timing (str, optional) – run_timing爲策略的運行時機，可以爲None，表示不指定運行時機
**kwargs (Any) – 添加的交易策略所共享的屬性，如 run_timing、run_freq、window_length、 use_latest_data_cycle、freq、asset_type、data_type_ids 等（會傳入 add_strategy）

回傳型別:

None

範例

>>> op = Operator()
>>> op.add_strategies(['dma', 'macd'])
>>> op.strategies
[RULE-ITER(DMA), RULE-ITER(MACD)]

add_strategy(stg: Union[str, BaseStrategy, type, tuple, list, Any], run_freq: str = 'd', run_timing: str = 'close', **kwargs)[原始碼]

添加一個strategy交易策略到operator對象中

如果調用本方法添加一個交易策略到Operator中，可以在添加的時候同時修改或指定交易策略的基本屬性

參數:

stg (str or int or Strategy) – 需要添加的交易策略，也可以是內置交易策略的策略id或策略名稱
run_freq (str, Optional) – run_freq爲策略的運行頻率，可以爲None，表示不指定運行頻率
run_timing (str, Optional) – run_timing爲策略的運行時機，可以爲None，表示不指定運行時機
kwargs – 任意合法的策略屬性，可以在添加策略時直接給該策略屬性賦值，必須明確指定需要修改的屬性名稱，包含 - pars: dict or tuple, 策略可調參數 - opt_tag: int, 策略優化標籤 - stg_type: int, 策略類型 - par_count: int, 策略參數個數 - par_types: list, 策略參數類型 - par_ranges: list, 策略參數範圍 - data_freq: str, 策略數據頻率 - window_length: int, 策略窗口長度 - run_freq: str, 策略採樣頻率 - data_types: list, 策略數據類型 - data_type_id: str, 策略數據類型 ID（用於 update_data_types） - freq: str or list or dict, 數據類型的頻率（修改會替換對應 DataType） - asset_type: str or list or dict, 數據類型的資產類型（修改會替換對應 DataType） - group: str, 策略運行時機 - use_latest_data_cycle: bool, 策略是否使用最新數據週期

回傳型別:

None

範例

>>> op = Operator()
>>> op.add_strategy('dma', opt_tag=1, pars=(50, 10, 20))
>>> op.strategies
[RULE-ITER(DMA)]
>>> op.strategies[0].opt_tag
1
>>> op.strategies[0].par_values
(50, 10, 20)

property all_dynamic_dtypes

返回 operator 內“舊式”動態數據類型集合；已廢棄，始終返回空。

過程數據現統一通過 get_data(『proc.xxx』) 訪問，不再通過 DataType 聲明。保留本屬性僅爲兼容調用方，返回空字典。

property all_strategy_data_types: 返回operator對象所有策略自對象的回測價格類型和交易清單歷史數據類型的集合

check_dynamic_data()[原始碼]

檢查operator對象是否包含動態數據類型（即依賴交易結果的歷史數據）以生成交易信號。

若任意策略在 realize() 中使用了 get_data(『proc.xxx』)，則視爲依賴動態過程數據，需走動態回測分支。

clear_strategies()[原始碼]: 清空Operator對象中的所有交易策略

create_data_windows()[原始碼]: Create data windows for each strategy and its data types. Also create data window indices for each strategy and its data types. data window indices are created according to group schedules.

disable_tracing()[原始碼]

禁用Operator對象中所有strategy的跟蹤功能

禁用跟蹤功能後，Operator對象在運行過程中不會記錄調試資訊，從而提高運行效率

回傳型別:: None

property empty: 檢查operator是否包含任何策略

enable_tracing()[原始碼]

啓用Operator對象中所有strategy的跟蹤功能

啓用跟蹤功能後，Operator對象在運行過程中會記錄更多的調試資訊，便於後續分析和調試

回傳型別:: None

get_blender(group_name=None)[原始碼]

返回operator對象中的多空濛板混合器, 如果不指定group_name的話，輸出完整的blender字典

參數:: group_name (str) – 一個可用的group_name
回傳:: blender – 如果group_name爲None，則返回一個字典，其中包含所有的run_timing的blender 如果group_name不爲None，則返回一個列表，其中包含該run_timing的blender
回傳型別:: dict or list

get_group(group_idx: Union[str, int]) → Group[原始碼]

根據group_idx獲取一個Group對象，等同於get_group_by_id方法

參數:: group_idx (int or str) – 策略組的序號
回傳型別:: Group,

範例

>>> op = Operator('dma, macd')
>>> op.get_group(0)
Group(name=Group_1, members=[RULE-ITER(DMA), RULE-ITER(MACD)])
>>> op.get_group_by_id('Group_1')
Group(name=Group_1, members=[RULE-ITER(DMA), RULE-ITER(MACD)])

get_group_by_id(group_id: Union[str, int]) → Group[原始碼]

根據group_id獲取一個Group對象

參數:: group_id (str or int) – 策略組的名稱ID或序號
回傳型別:: Group,

備註

1，當group_id爲int時，返回的是序號爲group_id的策略組 2，當group_id爲str時，返回的是ID爲group_id的組

範例

>>> op = Operator('dma, macd')
>>> op.get_group_by_id(0)
Group(name=Group_1, members=[RULE-ITER(DMA), RULE-ITER(MACD)])
>>> op.get_group_by_id('Group_1')
Group(name=Group_1, members=[RULE-ITER(DMA), RULE-ITER(MACD)])

get_max_window_length_by_dtype_id(dtype: str) → int[原始碼]

計算並返回operator對象某個datatype最長的窗口長度。

參數:: dtype (str) – 需要查詢的歷史數據類型的dtype_id
回傳型別:: int, operator对象中所有子策略中某个dtype最长的窗口长度

get_share_idx(share)[原始碼]

給定一個share（字符串）返回它對應的index

參數:: share (str) – share爲一個字符串，表示股票代碼
回傳:: 返回一個整數，表示share對應的index
回傳型別:: int

get_signal_count(steps=None) → int[原始碼]

獲取當前運行時間表中所有策略組生成的交易信號數量

參數:: steps (list of int, optional) – 如果給出steps，則只計算這些步驟對應的交易信號數量如果爲None，則計算所有步驟的交易信號數量
回傳:: int
回傳型別:: 交易信号的数量

get_stg(stg_id)[原始碼]

獲取一個strategy對象, Operator[item]的另一種用法

參數:: stg_id (int or str) – 策略的名稱或序號
回傳型別:: Strategy, 子策略

備註

1，當stg_id爲int時，返回的是第stg_id個策略 2，當stg_id爲str時，返回的是名稱爲stg_id的策略 3，當stg_id不符合要求時，返回最後一個策略

範例

>>> op = Operator('dma, macd')
>>> op[0]
RULE-ITER(DMA)
>>> op['dma']
RULE-ITER(DMA)
>>> op[999]
RULE-ITER(MACD)
>>> op['invalid_strategy_name']
UserWarning: No such strategy with ID (invalid_strategy_name)!

也參考

get_strategy_by_id

get_strategies_by_group(group_id: str)[原始碼]

返回operator對象中的strategy對象, timing爲一個可選參數，如果給出timing時，返回使用該timing的交易策略

參數:: group_id (str) – 一個可用的timing, by default None

get_strategy_by_id(stg_id)[原始碼]

獲取一個strategy對象, Operator[item]的另一種用法

參數:: stg_id (int or str) – 策略的名稱或序號
回傳型別:: Strategy, 子策略

也參考

get_stg

get_strategy_count_by_group(group_id: str)[原始碼]: 返回策略組group_id中的所有策略數量

get_strategy_id_by_group(group_id: str)[原始碼]: 返回策略組group_id中的所有策略ID

get_strategy_id_pairs()[原始碼]

返回一個generator，包含op中所有strategy和id對：

回傳型別:: generator, 包含op中所有strategy和id对

範例

>>> op = Operator('dma, macd')
>>> list(op.get_strategy_id_pairs())
[('dma', RULE-ITER(DMA)), ('macd', RULE-ITER(MACD))]

get_strategy_names_by_group(group_id: str)[原始碼]: 返回策略組group_id中的所有策略名稱

property group_ids: 返回operator對象所有策略子對象的運行時間類型

property group_merge_type: 返回operator對象的策略組合並方式

property group_names: 返回operator對象所有策略子對象的運行時間類型

property groups: 返回operator的所有策略組，返回以策略組的名稱爲索引的字典

property groups_by_index: 返回operator的所有策略組，返回以策略組的序號爲索引的字典

info(verbose=False)[原始碼]

列印Operator對象的資訊，包括策略組、組內策略，策略混合方式等等資訊

如果策略包含更多的資訊，還會列印出策略的一些具體資訊

參數:: verbose (bool, Default False) – 是否列印出策略的詳細資訊, 如果爲True, 則會列印出策略的詳細資訊

is_ready(tell_me_why: bool = False, raise_error: bool = False) → bool[原始碼]

檢查 Operator 是否已具備生成交易信號所需的最小條件。

典型就緒條件包括：已掛載策略、各策略組已配置 blender、歷史數據類型與運行時間表等關鍵字段已就緒（更細的合法性校驗在後續流程完成）。

參數:

tell_me_why (bool, default False) – 如果Operator對象不滿足準備好的條件，是否列印出具體原因, 默認不列印
raise_error (bool, default False) – 如果Operator對象不滿足準備好的條件，是否拋出異常, 默認不拋出

回傳型別:

bool, Operator对象是否已经准备好，可以开始生成交易信号

property max_window_length

計算並返回operator對象所有子策略中最長的窗口長度。在準備回測或優化歷史數據時，以此確保有足夠的歷史數據供策略形成

回傳型別:: int, operator对象中所有子策略中最长的窗口长度

property op_data_freq: dict: 返回operator對象所有策略子對象所需數據的採樣頻率如果所有strategy的data_freq相同時，給出這個值，否則給出一個排序的列表

property op_data_type_count: 返回operator對象生成交易清單所需的歷史數據類型數量

property op_data_type_ids: 返回operator對象所有策略子對象所需歷史數據類型的ID

property op_data_types: 返回operator對象所有策略子對象所需歷史數據類型對象

property op_signal_index: Optional[Index]: 返回operator對象生成交易信號的index

property op_signal_types: list

生成的交易清單的signal_types。因爲生成的交易信號可能是由不同的策略組生成的而不同的策略組有自己的信號類型（PT/ST/VT)，因此每一行交易信號都有可能有不同的交易信號類型，本屬性返回一個list，其中包含了op_signal_list中每一行的signal_list

回傳型別:: list, 生成的交易清单的price_types，回测交易价格类型

property op_type: 返回operator對象的運行類型

property opt_space_par

返回參與優化的子策略參數空間資訊，用於構造 Space。

遍歷各子策略的 opt_tag，彙總其 par_range / par_types 等字段，輸出 (ranges, types) 二元組，可直接作爲 Space 構造輸入。

回傳:: ranges 與 types 兩個列表，分別對應各參與優化維度的取值範圍與類型標記。
回傳型別:: tuple[list, list]

property opt_tags: 返回所有策略的優化類型標籤該屬性返回值是一個列表，按順序列出所有交易策略的優化類型標籤

prepare_data_buffer(*, start_date: Union[str, Timestamp], end_date: Union[str, Timestamp], data_package: dict) → None[原始碼]

準備數據緩衝區，加載所有策略需要的數據

數據緩衝區是一個字典，鍵爲數據類型，值爲對應的數據DataFrame，輸入參數包括數據包的開始和結束日期，根據這兩個日期從數據包中的每一個DataFrame中切片出相應的時間段，保存到數據緩衝區中。

保存數據緩衝時，還要檢查並確保數據有足夠的前置量以創建數據滑窗

參數:

start_date (str or pd.Timestamp) – 數據的開始日期，默認爲None，表示從數據包的起始日期開始
end_date (str or pd.Timestamp) – 數據的結束日期，默認爲None，表示到數據包的結束日期爲止
data_package (dict[]) – 一個字典，包含所有需要的數據，鍵爲數據類型，值爲對應的數據DataFrame 例如：{『price』: price_df, 『volume』: volume_df, …} 其中每個DataFrame的索引爲時間戳，列爲不同的標的代碼

prepare_dynamic_data_buffer(*, trade_records: ndarray, trade_prices: ndarray, own_cashes: ndarray, available_cashes: ndarray, holding_positions: ndarray, available_positions: ndarray) → None[原始碼]: 預留接口：過程數據已統一通過 proc.* 注入，本方法不再執行邏輯。

prepare_running_schedule(start_date=None, end_date=None, **kwargs) → None[原始碼]

Running Schedule也就是策略運行時間表，包含每個策略的運行時間和頻率等資訊

在運行策略之前，必須先準備好運行時間表，這個時間表根據交易員中每個策略組的運行時機參數確定。運行時間表包括N行，每一行代表一個時間點，列數爲策略組的數量，每個單元格表示該策略組在該時間點是否運行，0表示不運行，1表示運行。

在這個方法中，將設置以下兩個屬性的值： - group_schedules: 一個字典，鍵爲策略組名稱，值爲該組的運行時間表 - group_timing_table: 一個DataFrame，包含所有策略組的運行時間表

參數:

start_date (str or pd.Timestamp, optional) – 開始日期，默認爲None，表示從數據源的起始日期開始
end_date (str or pd.Timestamp, optional) – 結束日期，默認爲None，表示到數據源的結束日期爲止
kwargs (dict, optional) – 其他參數，包括生成交易時間序列過程中所需的參數，如交易開始時間、結束時間等等。詳見qteasy.trading_util.trade_time_index()函數的參數說明

回傳型別:

None

property ready: 屬性，operator.is_ready()的另一種寫法

remove_strategy(id_or_pos: Optional[Union[str, int]] = None) → None[原始碼]: 從Operator對象中移除一個交易策略, 刪除時可以給出策略的id或者策略在Operator中的位置

reset()[原始碼]

重置Operator對象的運行狀態，使其可以重新開始生成交易信號

備註

該方法會重置Operator對象的所有運行數據，包括歷史數據緩存、數據窗口視圖、數據窗口索引、交易信號等。重置後，Operator對象可以重新開始生成交易信號。

範例

>>> op = Operator('dma, macd')
>>> op.reset()

run(config, datasource=None, logger=None)[原始碼]

根據配置參數運行operator，支持實盤和回測兩種模式

參數:

config (dict) – 運行配置參數字典
datasource (DataSource, optional) – 數據源對象，默認爲None，表示使用全局默認數據源
logger (Logger, optional) – 日誌記錄器對象，默認爲None，表示不使用日誌記錄器

回傳:

backtest_result – 如果運行模式爲回測，則返回回測結果對象，包含回測的各種結果數據如果運行模式爲實盤，則不返回任何結果

回傳型別:

BacktestResult, optional

run_backtest(config, datasource=None, logger=None) → dict[原始碼]

在回測模式下運行operator, 使用歷史數據進行交易策略的回測並生成回測結果

參數:

config (dict) – 回測配置參數字典
datasource (DataSource, optional) – 數據源對象，默認爲None，表示使用全局默認數據源
logger (Logger, optional) – 日誌記錄器對象，默認爲None，表示不使用日誌記錄器

回傳:

backtest_result – 回測結果對象，包含回測的各種結果數據

回傳型別:

BacktestResult

run_live_trade(config, datasource=None, logger=None)[原始碼]: 在實盤模式下運行operator

run_optimization(config, datasource=None, logger=None) → dict[原始碼]: 在優化模式下運行operator

run_prediction(config, datasource=None, logger=None) → dict[原始碼]: 在與測模式下運行operator

run_strategies(steps: Iterable) → Iterable[原始碼]

運行 Operator，返回運行結果；語義接近 qt.run(self, ...) 傳入的關鍵字參數形式。

參數:: steps (Iterable) – 一個可迭代對象，包含需要運行的步驟索引
產出:: generator ((signal_type, step_index, signal)) – 返回一個生成器，包含每個步驟的交易信號 signal_type: str, 策略組的信號類型 step_index: int, 當前步驟的索引 signal: np.ndarray, 交易信號，一組數字，在不同信號類型模式下表示不同的含義

run_strategy(step_index) → Generator[Union[tuple[Any, int, Any], tuple[Optional[Any], int, Union[int, Any]]], Any, None][原始碼]

運行當前步驟的所有策略組，生成交易信號

本函數是一個生成器函數，返回每個策略組在當前步驟的交易信號。

參數:: step_index (int) – 當前步驟的索引，表示在運行時間表中的位置
回傳:: generator – 返回一個生成器，包含每個策略組在當前步驟的交易信號 signal_type: str, 策略組的信號類型 step_index: int, 當前步驟的索引 signal: np.ndarray, 交易信號，一組數字，在不同信號類型模式下表示不同的含義
回傳型別:: (signal_type, step_index, signal)

set_blender(blender: Union[str, list[str], dict[str, str]], group_id: Optional[str] = None)[原始碼]

統一的blender混合器屬性設置入口

參數:

blender (str or list of str or dict of str, optional) – 一個合法的交易信號混合表達式當group爲None時，可以接受list爲參數，同時爲所有的group設置混合表達式
group_id (str, optional) – 指定策略組 id 時僅更新該組；爲 None 時更新全部策略組。後者在 blender 爲字符串時表示所有組使用同一表達式；爲列表時按組順序逐項設置，長度不足則複用最後一項。

回傳型別:

None

引發:

TypeError – 如果給出的price_type不是正確的類型

警告

如果給出的price_type不存在，則給出warning並返回

範例

>>> op = Operator()
>>> op.add_strategy('dma', run_timing='close')  # 添加策略时指定run_timing，自动分配到第一个策略组Group_1
>>> op.add_strategy('trix', run_timing='close')  # 添加策略时指定run_timing，自动分配到第一个策略组Group_1
>>> op.add_strategy('macd', run_timing='open')  # 添加策略时指定run_timing，自动分配到第二个策略组Group_2
>>> op.add_strategy('bband', run_timing='open')  # 添加策略时指定run_timing，自动分配到第二个策略组Group_2

>>> # 设置策略组1的混合模式
>>> op.set_blender('s0+s1', 'Group_1')
>>> op.get_blender()
>>> {'Group_1': ['+', 's1', 's2']}

>>> # 给所有的交易价格策略设置同样的混合表达式
>>> op.set_blender('s0 + s1')
>>> op.get_blender()
>>> {'Group_1': ['+', 's2', 's1'], 'Group_2':  ['+', 's2', 's1']}

>>> # 通过一个列表给不同的策略组设置不同的混合表达式（策略组按顺序排列）
>>> op.set_blender(['s1 + s2', 's3*s4'], None)
>>> op.get_blender()
>>> {'close': ['+', 's2', 's1'], 'open':  ['*', 's4', 's3']}

set_group_parameters(group: Union[str, int], run_timing: Optional[str] = None, run_freq: Optional[str] = None, signal_type: Optional[str] = None, blender_str: Optional[str] = None, **kwargs)[原始碼]

設置或修改一個策略組的參數 :param group: 策略組的ID :type group: str :param run_timing: 策略組的運行時機，修改運行時機時，修改策略組中所有交易策略的運行時機 :type run_timing: str, optional :param run_freq: 策略組的運行頻率，修改運行頻率時，修改策略組 :type run_freq: str, optional :param signal_type: 策略組的交易信號類型，默認爲’PT』，即百分比持倉目標 :type signal_type: str, optional :param blender_str: 策略組的交易信號混合表達式，可以是一個字符串或一個字符串列表 :type blender_str: str, optional :param kwargs: 其他參數，可以是任意合法的策略組參數，如group_name, run_timing, run_freq等 :type kwargs: dict, optional

回傳型別:

None

引發:

TypeError – 如果group不是字符串或整數類型
ValueError – 如果group不存在或無法找到

set_opt_par_values(par_values)[原始碼]

優化器側入口：將一條參數向量按各子策略 opt_tag 切片寫回。

與 set_parameter 不同，本方法面向「一條向量對應多個策略子塊」的優化流程；調用方需保證 par_values 與 opt_space_par 展平順序一致。

參數:: par_values (tuple) – 優化器當前候選點對應的參數元組。
回傳型別:: None

備註

內部調用 Strategy.update_par_values，不在此處做參數合法性校驗。

set_parameter(stg_id: Union[str, int], pars: Optional[Union[Parameter, tuple, list, dict]] = None, opt_tag: Optional[int] = None, data_types: Optional[Union[DataType, list[qteasy.datatypes.DataType], dict[str, qteasy.datatypes.DataType]]] = None, data_type_id: Optional[str] = None, window_length: Optional[Union[int, tuple[int, ...], list[int], dict[str, int]]] = None, use_latest_data_cycle: Optional[Union[bool, list[bool], tuple[bool, ...], dict[str, bool]]] = None, freq: Optional[Union[str, list[str], tuple[str, ...], dict[str, str]]] = None, asset_type: Optional[Union[str, list[str], tuple[str, ...], dict[str, str]]] = None, par_values: Optional[Union[tuple, list, dict[str, Any]]] = None, par_range: Optional[Union[tuple, list, dict[str, tuple]]] = None, run_freq: Optional[str] = None, run_timing: Optional[str] = None, **kwargs)[原始碼]

統一的策略參數設置入口，stg_id標識接受參數的具體成員策略，將函數參數中給定的策略參數賦值給相應的策略

參數:

stg_id (str,) – 策略的名稱（ID），根據ID定位需要修改參數的策略
pars (tuple or dict, optional) – 可調策略參數，格式爲tuple 在創建一個策略的時候，可以設置部分策略參數爲」可調參數」，這些參數的取值範圍可以在策略優化過程中進行調整，通過調整這些參數的組合，可以找到最優的策略參數組合，從而找到最優的策略
opt_tag (int, optional) – 優化類型： 0: 不參加優化，在策略優化過程中不調整該策略的可調參數 1: 參加優化，在策略優化過程中根據優化算法主動調整策略參數以尋找最佳參數組合 2: 以枚舉類型參加優化，在策略優化過程中僅從給定的參數組合種選取最優的參數組合
data_types (DataType or list of DataType or dict of str, optional) – 策略計算所需歷史數據的數據類型，如果給出，則更新這個數據類型的參數
data_type_id (str, optional) – 策略計算所需歷史數據的數據類型的ID，給出該ID表明更新這個數據類型的參數
window_length (int or list of int or tuple of int,) – 窗口長度：策略計算的前視窗口長度
use_latest_data_cycle (bool or list of bool or tuple of bool,) – 是否使用最新的數據週期
freq (str or list of str or tuple of str or dict of str,) – 數據類型的頻率，若給出則用新 DataType 替換對應項（dtype_id 會隨之變化）
asset_type (str or list of str or tuple of str or dict of str,) – 數據類型的資產類型，若給出則用新 DataType 替換對應項（dtype_id 會隨之變化）
par_values (tuple or list,) – 策略參數的具體取值
par_range (tuple or list, or dict of tuples,) – 策略參數的取值範圍
run_freq (str, optional) – 如果給出該參數，則修改策略的運行頻率，修改運行頻率會導致將策略從策略組中移除，並重新分配到一個新的策略組中
run_timing (str, optional) – 如果給出該參數，則修改策略的運行時機，修改運行時機會導致將策略從策略組中移除，並重新分配到一個新的策略組中
kwargs (dict,) – 其他參數

set_shares(shares: list[str])[原始碼]

設置operator對象的交易標的列表

參數:: shares (list of str) – 一個字符串列表，表示交易標的代碼列表
回傳型別:: None

property strategies: 以列表的形式返回operator對象的所有Strategy對象

property strategy_count: 返回operator對象中的所有Strategy對象的數量

property strategy_group_count

計算operator對象中所有交易策略組的數量

回傳型別:: int, operator对象中所有交易策略组的数量

property strategy_groups: 返回operator的所有策略組，返回以策略組的名稱爲索引的字典

property strategy_ids: 返回operator對象中所有交易策略對象的ID

view_blender(group: Optional[Union[str, int]] = None) → Union[dict, str][原始碼]

返回operator對象中的多空濛板混合器的可讀版本, 即返回blender的原始字符串的更加可讀的: 版本，將s0等策略代碼替換爲策略ID，並進行適當格式化。如果不給出group參數，則返回所有策略組的blender可讀版本

參數:: group (str) – 一個可用的group的ID或index

Group 策略組類

class qteasy.group.Group(name: str, signal_type: str = 'pt', blender: Optional[str] = None, run_freq: str = 'd', run_timing: str = 'close')[原始碼]

策略組，包含多個策略成員，定義了成員策略的信號類型、混合方式、運行頻率和時機等屬性。

add_strategy(strategy: BaseStrategy) → None[原始碼]

添加一個策略到組中，策略將成爲該組的成員。

新增的策略必須是 BaseStrategy 的實例，並且不能已經是該組的成員，否則將引發 ValueError。新增策略的 group_id 將被設置爲該組的 group_id，以便在策略中訪問所屬組的資訊。

參數:

strategy (BaseStrategy) – 需要添加到組中的策略實例，必須是 BaseStrategy 的子類實例。

回傳型別:

None

引發:

ValueError – 當 strategy 已經是該組的成員時，拋出 ValueError，提示策略已存在於組中。
TypeError – 當 strategy 不是 BaseStrategy 的實例時，拋出 TypeError，提示策略類型錯誤。

blend(signals: Iterable)[原始碼]: 使用當前 blender 將同組策略信號混合爲一組信號。

clear_strategies()[原始碼]: 清空組中的所有策略，所有成員策略將不再是該組的成員。

property group_id: Group 的 ID，與 name 相同，用於 Strategy.group_id 兼容

remove_strategy(strategy: BaseStrategy)[原始碼]

從組中移除一個策略，策略將不再是該組的成員。

被移除的策略必須是該組的成員，否則將引發 ValueError。被移除策略的 group_id 將被設置爲 None，以表示不再屬於任何組。

參數:

strategy (BaseStrategy) – 需要從組中移除的策略實例，必須是該組的成員。

回傳型別:

None

引發:

ValueError – 當 strategy 不是該組的成員時，拋出 ValueError，提示策略不在組中。
TypeError – 當 strategy 不是 BaseStrategy 的實例時，拋出 TypeError，提示策略類型錯誤。