Repository: pdpipe/pdpipe

Issue:
Feature Request: Dynamic params from input dataframe on fit_transform
I want a way to supply pipeline stage constructor parameters with future-like placeholders, so that actual values will be determined by prior stages on application time.

I would assign the name of the future `ApplicationContext` key holding the value after it was calculated — wrapped by a unique class — to the parameter.

The constructor will hold on to this object, and will — in application time — pull the value of the right key from either the fit context or the application context (depending on how I set it: I might want this value to be set on pipeline fit, or to be set on each application dynamically, even on transforms when the pipeline is fitted), and use it for the transformation. Default should probably be the fit context?

Here's an example:

```python
import numpy as np; import pandas as pd; import pdpipe as pdp;

def scaling_decider(X: pd.DataFrame) -> str:
    """Determines with type of scaling to apply by examining all numerical columns."""
    numX = X.select_dtypes(include=np.number)
    for col in numX.columns:
        # this is nonsense logic, just an example
        if np.std(numX[col]) > 2 * np.mean(numX[col]):
            return 'StandardScaler'
    return 'MinMaxScaler'

pipeline = pdp.PdPipeline(stages=[
    pdp.ColDrop(pdp.cq.StartWith('n_'),
    pdp.Scale(
        # fit=False means it will take it from the application context, and not fit context
        scaler=pdp.dynamic(scaling_decider, fit=False),  
        joint=True,
    ),
])
```

## Design

This has to be implemented at the `PdPipeline` base class. Since the base class can't hijack constructor arguments, I think the contract with extending classes should be:

1. When implementing class exteding `PdPipeline`, if you want to enjoy support for contextual constructor parameters, you **MUST** delay any initialization of inner state objects to `fit_transform`, so that fit/application context is available on initialization (it is NOT available at pipeline stage costruction and initialization, after all).

2. `pdp.contextual` is a factory function that returns contextual parameter placeholder objects. Code using it shouldn't really care about it, as it should never interact with the resulting objects directly. I think.

3. `PdPipeline` can auto-magically make sure that any attribute of a `PdPipeline` instance that is assigned a `pdp.contextual` object in the constructor (e.g. `self.k = k`, and the `k` constructor argument was provided with `k=pdp.future('pca_k')`) will be hot-swapped with a concrete value by the time we wish to use it `fit_transform` or `transform` (for example, when we call `self.pca_ = PCA(k=self.k)`). It can also do so for any such object that is contained in any iterable or dict-like attribute (so if I have `self._pca_kwargs = {...}` in my constructor, I can safely call `self.pca_ = PCA(**self._pca_kwargs)` in `fit_transform()`.

## Implementation thoughts

To make this efficient, since this means accessing sub-class attribute instance on pipeline transformations, I have a few thoughts:

1. The `contextuals` module should have a global variable such as `CONTEXTUALS_ARE_ON = False`. Then, the `pdp.contextual` factory function sets `global CONTEXTUALS_ARE_ON; CONTEXTUALS_ARE_ON = True` when called. Then, we condition the whole inspection-heavy logic on this indicator variable, so that if our user never called `pdp.contextual` during the current kernel, runtime is saved.

2. I first thought `pdp.contextual` could somehow register `_ContextualParam` objects in a registery we could use to find what needed to be swapped, but actually this wouldn't help, as they won't know which attribute of which pipeline stage they were assigned to.

3. We thus have to scan sub-class attributes, but we can do so **if and only** after `pdp.contextual` was called, and right after pipeline stage initialization. Moreover, we can create a literal list of all attribute names we should ignore, stored in `pdpipe.core` as a global: e.g. `_IGNORE_PDPSTAGE_ATT = ['_desc', '_name']`, etc. Everything we know isn't an attribute the sub-class declared. Then, we can check any attribute that isn't one of these. This can be done in `pdpipe.PdPipelineStage.__init__()`, since it's called (by contract; we can demand that from extending subclasses) at the end of the `__init__()` method of subclasses. When we find that such an attribute holds a `pdp.contextual`, we register it at the `pdp.contextuals` module, in some global dict (or something more sophisticated), keyed by the attribute name. We can also registed the containig stage object.

Then, in a stage `fit_transform` and `transform` methods, if the current stage object is registered for contextual hot-swapping, we find the concrete contextual value of any attribute resigtered for this stage (in either the `self.fit_context` object or `self.application_context` object the pipeline injects all stages during applications of the pipeline) and hot-swap it literaly: This will look something like `setattr(self, 'k', self.fit_context['pca_k'])`, since we're at `pdpipe.PdPipelineStage.fit_transform()`, and the `self` object is an instance of the subclass requiring the hot swap (in this case, `pdp.Decompose`).

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/pdpipe/pdpipe/pull/111?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) Report
> Merging [#111](https://codecov.io/gh/pdpipe/pdpipe/pull/111?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (c809e68) into [master](https://codecov.io/gh/pdpipe/pdpipe/commit/239f9cf6b747459f11fd9810b996082279b2607d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) (239f9cf) will **increase** coverage by `0.02%`.
> The diff coverage is `98.24%`.

[![Impacted file tree graph](https://codecov.io/gh/pdpipe/pdpipe/pull/111/graphs/tree.svg?width=650&height=150&src=pr&token=SB4uE4HWbv&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)](https://codecov.io/gh/pdpipe/pdpipe/pull/111?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)

```diff
@@            Coverage Diff             @@
##           master     #111      +/-   ##
==========================================
+ Coverage   98.04%   98.06%   +0.02%     
==========================================
  Files          24       25       +1     
  Lines        3625     3671      +46     
  Branches      646      652       +6     
==========================================
+ Hits         3554     3600      +46     
  Misses         56       56              
  Partials       15       15              
```


| [Impacted Files](https://codecov.io/gh/pdpipe/pdpipe/pull/111?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None) | Coverage Δ | |
|---|---|---|
| [pdpipe/pdp\_types.py](https://codecov.io/gh/pdpipe/pdpipe/pull/111/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGRwaXBlL3BkcF90eXBlcy5weQ==) | `100.00% <ø> (ø)` | |
| [pdpipe/core.py](https://codecov.io/gh/pdpipe/pdpipe/pull/111/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGRwaXBlL2NvcmUucHk=) | `90.67% <96.15%> (+0.26%)` | :arrow_up: |
| [pdpipe/basic\_stages.py](https://codecov.io/gh/pdpipe/pdpipe/pull/111/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGRwaXBlL2Jhc2ljX3N0YWdlcy5weQ==) | `99.71% <100.00%> (ø)` | |
| [pdpipe/col\_generation.py](https://codecov.io/gh/pdpipe/pdpipe/pull/111/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGRwaXBlL2NvbF9nZW5lcmF0aW9uLnB5) | `100.00% <100.00%> (ø)` | |
| [pdpipe/df/bound\_column\_potential.py](https://codecov.io/gh/pdpipe/pdpipe/pull/111/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGRwaXBlL2RmL2JvdW5kX2NvbHVtbl9wb3RlbnRpYWwucHk=) | `100.00% <100.00%> (ø)` | |
| [pdpipe/run\_time\_parameters.py](https://codecov.io/gh/pdpipe/pdpipe/pull/111/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGRwaXBlL3J1bl90aW1lX3BhcmFtZXRlcnMucHk=) | `100.00% <100.00%> (ø)` | |
| [pdpipe/shared.py](https://codecov.io/gh/pdpipe/pdpipe/pull/111/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGRwaXBlL3NoYXJlZC5weQ==) | `100.00% <100.00%> (ø)` | |
| [pdpipe/text\_stages.py](https://codecov.io/gh/pdpipe/pdpipe/pull/111/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None#diff-cGRwaXBlL3RleHRfc3RhZ2VzLnB5) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pdpipe/pdpipe/pull/111?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pdpipe/pdpipe/pull/111?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None). Last update [239f9cf...c809e68](https://codecov.io/gh/pdpipe/pdpipe/pull/111?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=None).


Patch:
diff --git a/.readthedocs.mkdocs.yml b/.readthedocs.mkdocs.yml
index fa4d801..d3cf57d 100644
--- a/.readthedocs.mkdocs.yml
+++ b/.readthedocs.mkdocs.yml
@@ -41,6 +41,7 @@ nav:
       - Conditions: reference/cond.md
       - Column Qualifiers: reference/cq.md
       - Wrappers: reference/wrappers.md
+      - Run time parameters: reference/runtime_parameters.md
 
 theme:
   name: material
diff --git a/docs/reference/runtime_parameters.md b/docs/reference/runtime_parameters.md
new file mode 100644
index 0000000..108e4da
--- /dev/null
+++ b/docs/reference/runtime_parameters.md
@@ -0,0 +1,3 @@
+# Run time parameters
+
+::: pdp.run_time_parameters
diff --git a/docs/tutorials/runtime_parameters.md b/docs/tutorials/runtime_parameters.md
new file mode 100644
index 0000000..f3c2913
--- /dev/null
+++ b/docs/tutorials/runtime_parameters.md
@@ -0,0 +1,35 @@
+# Dynamic Runtime Parameters
+In some cases one wishes to set a parameter based on calculations done
+in real time, based on the given input. Dynamic parameters help to achieve 
+this with the `pdp.dynamic` function. The function is provided with a callable
+which implements the logic on deciding the parameter, and is applied only when
+the input is available.
+
+## Example
+The `scaling_decider` function implements the logic for choosing the parameter
+type (`'StandardScaler'` or `'MinMaxScaler'`) and is passed to the 
+`pdp.dynamic` function in the stage's (`Scale`) constructor.
+The logic references the given input and chooses a parameter based on it. 
+
+```python
+import numpy as np; import pandas as pd; import pdpipe as pdp;
+
+def scaling_decider(X: pd.DataFrame) -> str:
+    """
+    Determines which type of scaling to apply by examining all numerical 
+    columns.
+    """
+    numX = X.select_dtypes(include=np.number)
+    for col in numX.columns:
+        if np.std(numX[col]) > 2 * np.mean(numX[col]):
+            return 'StandardScaler'
+    return 'MinMaxScaler'
+
+pipeline = pdp.PdPipeline(stages=[
+    pdp.ColDrop(pdp.cq.StartWith('n_')),  # not connected to scale
+    pdp.Scale(
+        scaler=pdp.dynamic(scaling_decider, fit=False),  
+        joint=True,
+    )
+])
+```
diff --git a/pdpipe/__init__.py b/pdpipe/__init__.py
index cb4c0d5..9e8d719 100644
--- a/pdpipe/__init__.py
+++ b/pdpipe/__init__.py
@@ -120,6 +120,9 @@ except ImportError:
         "stages will not be loaded."
     )
 
+from .run_time_parameters import (
+    dynamic
+)
 
 from .df import DF_HANDLE as df
 
diff --git a/pdpipe/basic_stages.py b/pdpipe/basic_stages.py
index 64eb423..d5a41e9 100644
--- a/pdpipe/basic_stages.py
+++ b/pdpipe/basic_stages.py
@@ -14,7 +14,7 @@ from pdpipe.shared import (
 )
 
 import pdpipe.cond as cond
-from pdpipe.types import ColumnsParamType
+from pdpipe.pdp_types import ColumnsParamType
 from pdpipe.exceptions import FailedConditionError
 from pdpipe.cq import ColumnQualifier
 
diff --git a/pdpipe/col_generation.py b/pdpipe/col_generation.py
index 86045bb..790cbf4 100644
--- a/pdpipe/col_generation.py
+++ b/pdpipe/col_generation.py
@@ -18,7 +18,7 @@ from pdpipe.core import (
 )
 from pdpipe.util import out_of_place_col_insert
 from pdpipe.cq import OfDtypes
-from pdpipe.types import ColumnsParamType, ColumnLabelsType
+from pdpipe.pdp_types import ColumnsParamType, ColumnLabelsType
 
 from pdpipe.shared import (
     _interpret_columns_param,
diff --git a/pdpipe/core.py b/pdpipe/core.py
index 40cc46f..f657cd4 100644
--- a/pdpipe/core.py
+++ b/pdpipe/core.py
@@ -11,6 +11,9 @@ from typing import Tuple, Union, Iterable, Optional
 
 import numpy
 import pandas
+import pandas as pd
+
+from .run_time_parameters import DynamicParameter
 
 try:
     from pympler.asizeof import asizeof
@@ -22,6 +25,7 @@ from .cfg import (
 )
 from .cq import is_fittable_column_qualifier, AllColumns
 from .shared import (
+    POS_ARG_MISMTCH_PAT,
     _get_args_list,
     _always_true,
 )
@@ -284,6 +288,8 @@ class PdPipelineStage(abc.ABC):
         self.is_fitted = False
         self._is_being_applied = False
         self._is_being_fitted = False
+        self._dynamics = []  # list of parameters to be decided at runtime
+        self._process_dynamics()
 
     def is_being_fitted_by_pipeline(self) -> bool:
         """Returns True if this stage is being fitted."""
@@ -292,6 +298,25 @@ class PdPipelineStage(abc.ABC):
         except AttributeError:
             return False
 
+    class_attrs = {'_exraise', '_exmsg', '_exmsg_post',
+                   '_desc', '_prec_arg', '_post_arg', '_skip',
+                   '_appmsg', '_name', '_is_an_Xy_transformer',
+                   '_is_an_Xy_fit_transformer', 'fit_context',
+                   'application_context', 'is_fitted', '_dynamics'}
+
+    def _process_dynamics(self):
+        """
+        Creates a list of Dynamic attributes
+        Returns None
+        -------
+        """
+        potential_dynamics_attrs = set(self.__dict__).\
+            difference(self.class_attrs)
+        for attr in potential_dynamics_attrs:
+            attr_obj = self.__getattribute__(attr)
+            if isinstance(attr_obj, DynamicParameter):
+                self._dynamics.append({'name': attr, 'callable': attr_obj})
+
     @classmethod
     def _init_kwargs(cls):
         return cls._INIT_KWARGS
@@ -318,9 +343,6 @@ class PdPipelineStage(abc.ABC):
             k: v for k, v in kwargs.items() if k not in cls._INIT_KWARGS}
         return init_kwargs, other_kwargs
 
-    _POS_ARG_MISMTCH_PAT = re.compile(
-        r'\d positional argument[s]? but \d (were|was) given')
-
     _MISSING_POS_ARG_PAT = re.compile(
         r'missing \d+ required positional argument')
 
@@ -364,8 +386,7 @@ class PdPipelineStage(abc.ABC):
             try:
                 return to_call(X, y)
             except TypeError as e:
-                if len(PdPipelineStage._POS_ARG_MISMTCH_PAT.findall(
-                        str(e))) > 0:
+                if len(POS_ARG_MISMTCH_PAT.findall(str(e))) > 0:
                     return to_call(X)
                 raise e  # pragma: no cover
 
@@ -383,7 +404,7 @@ class PdPipelineStage(abc.ABC):
         try:
             return self._prec(X, y)
         except TypeError as e:
-            if len(PdPipelineStage._POS_ARG_MISMTCH_PAT.findall(str(e))) > 0:
+            if len(POS_ARG_MISMTCH_PAT.findall(str(e))) > 0:
                 return self._prec(X)
             raise e  # pragma: no cover
 
@@ -431,8 +452,7 @@ class PdPipelineStage(abc.ABC):
             try:
                 return to_call(X, y)
             except TypeError as e:
-                if len(PdPipelineStage._POS_ARG_MISMTCH_PAT.findall(
-                        str(e))) > 0:
+                if len(POS_ARG_MISMTCH_PAT.findall(str(e))) > 0:
                     return to_call(X)
                 raise e  # pragma: no cover
 
@@ -443,7 +463,7 @@ class PdPipelineStage(abc.ABC):
         try:
             return self._post(X, y)
         except TypeError as e:
-            if len(PdPipelineStage._POS_ARG_MISMTCH_PAT.findall(str(e))) > 0:
+            if len(POS_ARG_MISMTCH_PAT.findall(str(e))) > 0:
                 return self._post(X)
             raise e  # pragma: no cover
 
@@ -589,8 +609,7 @@ class PdPipelineStage(abc.ABC):
             try:
                 return self._skip(X, y)
             except TypeError as e:
-                if len(PdPipelineStage._POS_ARG_MISMTCH_PAT.findall(
-                        str(e))) > 0:
+                if len(POS_ARG_MISMTCH_PAT.findall(str(e))) > 0:
                     return self._skip(X)
                 else:
                     raise e  # pragma: no cover
@@ -1058,7 +1077,7 @@ class AdHocStage(PdPipelineStage):
         try:
             return self._adhoc_prec(X, y)
         except TypeError as e:
-            if len(PdPipelineStage._POS_ARG_MISMTCH_PAT.findall(str(e))) > 0:
+            if len(POS_ARG_MISMTCH_PAT.findall(str(e))) > 0:
                 return self._adhoc_prec(X)
             raise e
 
@@ -1177,15 +1196,38 @@ class PdPipeline(PdPipelineStage, collections.abc.Sequence):
         self.application_context = None
         self.fit_context.lock()
 
+    @staticmethod
+    def _use_dynamics(stage: PdPipelineStage,
+                      inter_X: pd.DataFrame,
+                      inter_y: Optional[pd.Series] = None) -> None:
+        """
+        sets the dynamic parameter based on given callable
+        Parameters
+        ----------
+        stage: stage for which to provide the dynamic parameter
+        inter_X: input df
+        inter_y: optional input y labels
+
+        Returns
+        -------
+
+        """
+        for dynamic in stage._dynamics:
+            param = dynamic['callable'](inter_X) if inter_y is None \
+                else dynamic['callable'](inter_X, inter_y)
+            setattr(stage,
+                    dynamic['name'],
+                    param)
+
     def apply(
-        self,
-        X: pandas.DataFrame,
-        y: Optional[pandas.Series] = None,
-        exraise: Optional[bool] = None,
-        verbose: Optional[bool] = False,
-        time: Optional[bool] = False,
-        fit_context: Optional[dict] = {},
-        application_context: Optional[dict] = {},
+            self,
+            X: pandas.DataFrame,
+            y: Optional[pandas.Series] = None,
+            exraise: Optional[bool] = None,
+            verbose: Optional[bool] = False,
+            time: Optional[bool] = False,
+            fit_context: Optional[dict] = {},
+            application_context: Optional[dict] = {},
     ):
         """Applies this pipeline stage to the given dataframe.
 
@@ -1377,6 +1419,8 @@ class PdPipeline(PdPipelineStage, collections.abc.Sequence):
                 try:
                     stage.fit_context = self.fit_context
                     stage.application_context = self.application_context
+                    self._use_dynamics(stage, inter_X, inter_y)
+
                     inter_X = stage.fit_transform(
                         X=inter_X,
                         y=None,
@@ -1394,6 +1438,8 @@ class PdPipeline(PdPipelineStage, collections.abc.Sequence):
                 try:
                     stage.fit_context = self.fit_context
                     stage.application_context = self.application_context
+                    self._use_dynamics(stage, inter_X, inter_y)
+
                     inter_X, inter_y = stage.fit_transform(
                         X=inter_X,
                         y=inter_y,
@@ -1590,6 +1636,8 @@ class PdPipeline(PdPipelineStage, collections.abc.Sequence):
                 try:
                     stage.fit_context = self.fit_context
                     stage.application_context = self.application_context
+                    self._use_dynamics(stage, inter_X, inter_y)
+
                     inter_X = stage.transform(
                         X=inter_X,
                         y=None,
@@ -1607,6 +1655,8 @@ class PdPipeline(PdPipelineStage, collections.abc.Sequence):
                 try:
                     stage.fit_context = self.fit_context
                     stage.application_context = self.application_context
+                    self._use_dynamics(stage, inter_X, inter_y)
+
                     inter_X, inter_y = stage.transform(
                         X=inter_X,
                         y=inter_y,
diff --git a/pdpipe/df/bound_column_potential.py b/pdpipe/df/bound_column_potential.py
index 2d277e4..617ba75 100644
--- a/pdpipe/df/bound_column_potential.py
+++ b/pdpipe/df/bound_column_potential.py
@@ -8,7 +8,7 @@ from pandas import DataFrame, Series
 
 from ..shared import _list_str
 from ..core import PdPipelineStage
-from ..types import SeriesOperandTypesTuple
+from ..pdp_types import SeriesOperandTypesTuple
 from .func_lists import (
     SERIES_TRANSFORMS_BLACKLIST,
     SERIES_TRANSFORMS_WHITELIST,
diff --git a/pdpipe/types.py b/pdpipe/pdp_types.py
similarity index 100%
rename from pdpipe/types.py
rename to pdpipe/pdp_types.py
diff --git a/pdpipe/run_time_parameters.py b/pdpipe/run_time_parameters.py
new file mode 100644
index 0000000..f3f20d7
--- /dev/null
+++ b/pdpipe/run_time_parameters.py
@@ -0,0 +1,85 @@
+from typing import Any, Callable, Optional
+import pandas as pd
+
+from .shared import POS_ARG_MISMTCH_PAT
+
+
+class DynamicParameter:
+    """
+    This class represents a dynamic parameter that is decided at fit time when
+    there is access to the given dataframe. This allows the parameter to be
+    decided at "run time" depending on the given dataframe
+
+    Parameters
+    ----------
+    parameter_selector: Callable, preforms logic in order to decide on a given
+     parameter at run time, using the given DataFrame and optionally y label
+     series.
+    fittable: boolean denoting whether the parameter can be fitted, or should
+     perform the deciding logic every time.
+    """
+    def __init__(
+            self,
+            parameter_selector: Callable,
+            fittable: Optional[bool] = True,  # if False, fit in each use
+    ) -> None:
+        self._parameter_selector = parameter_selector
+        self._fittable = fittable
+        self._parameter: Any
+
+    def fit_transform(self,
+                      X: pd.DataFrame,
+                      y: Optional[pd.Series] = None):
+        """
+
+        Parameters
+        ----------
+        X: Input dataframe upon which the parameter selector decides which
+           parameter to use
+        y: optional label column, if separated from X dataframe at an earlier
+           stage
+
+        Returns
+        -------
+        """
+        try:
+            self._parameter = self._parameter_selector(X, y)
+        except TypeError as e:
+            if len(POS_ARG_MISMTCH_PAT.findall(str(e))) > 0:
+                self._parameter = self._parameter_selector(X)
+            else:
+                raise TypeError('Parameter selector must be callable') from e
+        return self._parameter
+
+    def transform(self):
+        """
+        Transform returns the fitted parameter based on given dataframe
+        """
+        if not hasattr(self, '_parameter'):
+            raise AttributeError('Parameter not fitted')  # pragma: no cover
+        return self._parameter
+
+    def __call__(self,
+                 X: pd.DataFrame,
+                 y: Optional[pd.Series] = None,
+                 *args, **kwargs):
+        if self._fittable and hasattr(self, '_parameter'):
+            return self.transform()
+        else:
+            return self.fit_transform(X, y)
+
+
+def dynamic(parameter_selector: Callable, fittable: Optional[bool] = True) \
+        -> DynamicParameter:
+    """
+    Factory for the DynamicParameter class. This is used
+    Parameters
+    ----------
+    parameter_selector: callable holding the logic on how to select a certain
+     parameter based on given df
+    fittable: boolean denoting whether the parameter can be fitted, or should
+     perform the deciding logic every time.
+    Returns DynamicParameter object with the parameter selecting logic
+    -------
+    """
+    return DynamicParameter(parameter_selector, fittable)
diff --git a/pdpipe/shared.py b/pdpipe/shared.py
index e45b35f..d85f1d6 100644
--- a/pdpipe/shared.py
+++ b/pdpipe/shared.py
@@ -1,8 +1,12 @@
 """Shared inner functionalities for pdpipe."""
 
 import inspect
+import re
 from typing import List, Iterable
 
+POS_ARG_MISMTCH_PAT = re.compile(
+    r'\d positional argument[s]? but \d (were|was) given')
+
 
 def _interpret_columns_param(columns: object) -> List[object]:
     if isinstance(columns, str):
diff --git a/pdpipe/text_stages.py b/pdpipe/text_stages.py
index def8f6c..9123abc 100644
--- a/pdpipe/text_stages.py
+++ b/pdpipe/text_stages.py
@@ -4,7 +4,7 @@ import re
 from typing import Optional
 
 from pdpipe.col_generation import ApplyByCols
-from pdpipe.types import ColumnsParamType, ColumnLabelsType
+from pdpipe.pdp_types import ColumnsParamType, ColumnLabelsType
 
 
 class RegexReplace(ApplyByCols):


Test patch:
diff --git a/tests/run_time_parameters/test_dynamics.py b/tests/run_time_parameters/test_dynamics.py
new file mode 100644
index 0000000..bb6060e
--- /dev/null
+++ b/tests/run_time_parameters/test_dynamics.py
@@ -0,0 +1,118 @@
+import pandas as pd
+import pytest
+
+import pdpipe as pdp
+from pdpipe.exceptions import PipelineApplicationError
+
+
+def _standard_scaler_df():
+    return pd.DataFrame([[2, 3],
+                         [5, 6]],
+                        columns=['a', 'b'])
+
+
+def _minmax_scaler_df():
+    return pd.DataFrame([[2, 3],
+                         [5, 7]],
+                        columns=['a', 'b'])
+
+
+def _scaling_decider_no_y(X: pd.DataFrame) -> str:
+    """
+    Determines with type of scaling to apply by examining all numerical
+    columns.
+    """
+    for col in X.columns:
+        print(X[col].sum())
+        if X[col].sum() % 3 == 0:
+            return 'StandardScaler'
+    return 'MinMaxScaler'
+
+
+def _scaling_decider_with_y(X: pd.DataFrame, y) -> str:
+    """
+    Determines with type of scaling to apply by examining all
+    numerical columns.
+    """
+    if y.sum() % 2 == 0:
+        return 'MinMaxScaler'
+    return 'StandardScaler'
+
+
+def _get_scaler(param_selector, fittable: bool = False):
+    # fit=False, it will take it from the application context, not fit context
+    return pdp.Scale(
+        scaler=pdp.dynamic(param_selector, fittable=fittable),
+        joint=True,
+    )
+
+
+def test_scaler_with_y():
+    scaler = _get_scaler(_scaling_decider_with_y)
+    pipeline = pdp.PdPipeline(stages=[scaler])
+    df_standard_scaler = _standard_scaler_df()
+    df_standard_scaler_x = df_standard_scaler['a'].to_frame()
+    df_standard_scaler_y = df_standard_scaler['b']
+    res_x, res_y = pipeline.fit_transform(df_standard_scaler_x,
+                                          df_standard_scaler_y)
+    expected_x = pd.DataFrame([-1, 1], columns=['a'])
+    expected_y = pd.Series([3, 6])
+
+    assert scaler.scaler == 'StandardScaler'
+    pd.testing.assert_frame_equal(res_x, expected_x, check_dtype=False)
+    pd.testing.assert_series_equal(res_y, expected_y)
+
+    df_minmax_scaler = _minmax_scaler_df()
+    df_minmax_scaler_x = df_minmax_scaler['a'].to_frame()
+    df_minmax_scaler_y = df_minmax_scaler['b']
+    res_x, res_y = pipeline.fit_transform(X=df_minmax_scaler_x,
+                                          y=df_minmax_scaler_y)
+    expected_x = pd.DataFrame([0, 1], columns=['a'])
+    expected_y = pd.Series([3, 7])
+    pd.testing.assert_series_equal(res_y, expected_y)
+
+    assert scaler.scaler == 'MinMaxScaler'
+    pd.testing.assert_frame_equal(res_x, expected_x, check_dtype=False)
+    pd.testing.assert_series_equal(res_y, expected_y)
+
+
+def test_scaler_no_y():
+    scaler = _get_scaler(_scaling_decider_no_y)
+    pipeline = pdp.PdPipeline(stages=[scaler])
+    res = pipeline.apply(_standard_scaler_df())
+    expected_res = pd.DataFrame([[-1.264911, -0.632456],
+                                 [0.632456, 1.264911]], columns=['a', 'b'])
+
+    assert scaler.scaler == 'StandardScaler'
+    pd.testing.assert_frame_equal(res, expected_res)
+
+    pipeline.fit(_minmax_scaler_df())
+    res = pipeline.apply(_minmax_scaler_df())
+    expected_res = pd.DataFrame([[0, 0.2],
+                                 [0.6, 1]],
+                                columns=['a', 'b'])
+
+    assert scaler.scaler == 'MinMaxScaler'
+    pd.testing.assert_frame_equal(res, expected_res)
+
+
+def test_non_callable():
+    with pytest.raises(PipelineApplicationError):
+        scaler = _get_scaler('non_callable')
+        pipeline = pdp.PdPipeline(stages=[scaler])
+        pipeline.apply(_minmax_scaler_df())
+
+
+def test_fittable():
+    """ scaler should not change second time since the stage was fitted """
+    scaler = _get_scaler(_scaling_decider_no_y, fittable=True)
+    pipeline = pdp.PdPipeline(stages=[scaler])
+    pipeline.apply(_minmax_scaler_df())
+    assert scaler.scaler == 'MinMaxScaler'
+
+    # check that minmax scaler was applied to standard scaler df
+    res = pipeline.apply(_standard_scaler_df())
+    expected_res = pd.DataFrame([[0, 0.2],
+                                 [0.6, 0.8]],
+                                columns=['a', 'b'])
+    pd.testing.assert_frame_equal(res, expected_res)