Skip to content

create_all

Intuição

Quando você está explorando a biblioteca ou construindo um dataset para ML, é comum querer gerar várias features de uma vez.

create_all faz isso automaticamente: ela percorre as features públicas do quantmaster.features (via __all__), calcula cada feature com parâmetros default quando possível e adiciona o resultado ao DataFrame.

Comportamento

  • Retorna um novo DataFrame por padrão (inplace=False).
  • Não sobrescreve colunas existentes por padrão (overwrite=False).
  • Features que exigem argumentos extras (ex.: benchmark) são puladas se você não fornecer.
  • Features com dependências opcionais (ex.: path_signature_features) podem ser puladas automaticamente.
  • Por padrão, executa somente features com status approved.
  • Você pode controlar o comportamento com include, exclude, include_statuses e errors.

Uso

from quantmaster.features.utils import create_all

# adiciona todas as features possíveis com defaults
# (as que dependem de benchmark/dependência opcional podem ser puladas)
df = create_all(df)

# com benchmark para habilitar betas e spread_zscore
# df = create_all(df, benchmark=spx_close)

# gerar apenas um subconjunto
# df = create_all(df, include=["rsi", "har_rv"])

# incluir também features experimentais
# df = create_all(df, include_statuses=["approved", "experimental"])

API

Source code in src/quantmaster/features/utils.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def create_all(
    data: pd.DataFrame,
    *,
    benchmark: pd.Series | None = None,
    inplace: bool = False,
    overwrite: bool = False,
    include: Iterable[str] | None = None,
    exclude: Iterable[str] | None = None,
    include_statuses: Iterable[str] | None = None,
    errors: str = "ignore",
) -> pd.DataFrame:
    if not isinstance(data, pd.DataFrame):
        raise TypeError(f"data must be a pandas.DataFrame, got {type(data).__name__}")
    if errors not in {"ignore", "warn", "raise"}:
        raise ValueError(f"errors must be one of 'ignore', 'warn', 'raise', got {errors!r}")

    if inplace:
        out_df = data
    else:
        out_df = data.copy()

    include_set = set(include) if include is not None else None
    exclude_set = set(exclude) if exclude is not None else set()
    exclude_set |= DEFAULT_CREATE_ALL_EXCLUDE
    status_filter = normalize_status_filter(include_statuses)

    from quantmaster import features as _features

    names = [n for n in getattr(_features, "__all__", []) if isinstance(n, str)]

    for name in names:
        if include_set is not None and name not in include_set:
            continue
        if name in exclude_set:
            continue
        if status_filter is not None and get_feature_status(name) not in status_filter:
            continue

        fn = getattr(_features, name, None)
        if not callable(fn):
            continue

        try:
            sig = inspect.signature(fn)
        except (TypeError, ValueError):
            continue

        params = sig.parameters
        kwargs: dict[str, object] = {}
        if "benchmark" in params:
            p = params["benchmark"]
            if benchmark is None and p.default is inspect._empty:
                continue
            if benchmark is not None:
                kwargs["benchmark"] = benchmark

        try:
            res = fn(out_df, **kwargs)
        except Exception as exc:
            if errors == "raise":
                raise
            if errors == "warn":
                warnings.warn(f"create_all skipped '{name}': {exc}", RuntimeWarning)
            continue

        if isinstance(res, pd.Series):
            if res.isna().all():
                continue
            col = res.name or name
            if (not overwrite) and (col in out_df.columns):
                continue
            out_df[col] = res
            continue

        if isinstance(res, pd.DataFrame):
            if res.isna().all(axis=None):
                continue
            for c in res.columns:
                if (not overwrite) and (c in out_df.columns):
                    continue
                out_df[c] = res[c]
            continue

        if errors == "raise":
            raise TypeError(f"Feature '{name}' returned unsupported type {type(res).__name__}")
        if errors == "warn":
            warnings.warn(
                f"create_all skipped '{name}': unsupported return type {type(res).__name__}",
                RuntimeWarning,
            )

    return out_df