Skip to content

Schema

The adapter layer that converts ISO-specific trade DataFrames into the canonical format expected by the metrics pipeline. Each ISO registers an ISOTradeSchema instance that declares its column mapping, gain-normalization function, and optional validation hooks.

Adaptation Steps

  1. Validate that all required ISO columns exist.
  2. Rename ISO columns to canonical names using the inverse of column_map.
  3. Enforce datetime dtype on interval_start_local.
  4. Compute gain_normalized via the registered gain_normalizer callable.
  5. Derive date from interval_start_local.
  6. Add a constant iso column.
  7. Run any extra_validations.

ISO Trade Schema - Adapter layer for converting ISO-specific DataFrames to canonical format.

Classes

ISOTradeSchema dataclass 

ISOTradeSchema(iso_name: str, column_map: Mapping[str, str], gain_normalizer: GainNormalizer, capital_requirement_fn: CapitalRequirementFn | None = None, extra_validations: list[Callable[[DataFrame], None]] = list())

Schema definition for adapting ISO-specific trade DataFrames to canonical format.

Attributes: iso_name: Name of the ISO (e.g., "MISO", "ERCOT", "SPP") column_map: Mapping from canonical column names to ISO-specific column names gain_normalizer: Callable that computes gain_normalized from the DataFrame capital_requirement_fn: Optional callable to compute capital requirement extra_validations: List of validation functions to run on the adapted DataFrame

Functions

adapt_frame 
adapt_frame(df_iso: DataFrame) -> DataFrame

Adapt an ISO-specific DataFrame to the canonical format.

Steps: 1. Validate required columns exist (using self.column_map.values()) 2. Rename ISO columns to canonical names (inverse mapping) 3. Enforce dtypes (datetime64[ns], booleans, floats) 4. Create 'gain_normalized' via self.gain_normalizer(df_renamed) 5. Add 'date' = df_renamed["interval_start_local"].dt.date 6. Add constant column 'iso' = self.iso_name 7. Run any extra_validations 8. Return a new DataFrame (do not mutate input)

Args: df_iso: ISO-specific DataFrame

Returns: Standardized DataFrame with canonical columns

Source code in src/progridpy/metrics/schema.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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
def adapt_frame(self, df_iso: pd.DataFrame) -> pd.DataFrame:
    """
    Adapt an ISO-specific DataFrame to the canonical format.

    Steps:
    1. Validate required columns exist (using self.column_map.values())
    2. Rename ISO columns to canonical names (inverse mapping)
    3. Enforce dtypes (datetime64[ns], booleans, floats)
    4. Create 'gain_normalized' via self.gain_normalizer(df_renamed)
    5. Add 'date' = df_renamed["interval_start_local"].dt.date
    6. Add constant column 'iso' = self.iso_name
    7. Run any extra_validations
    8. Return a new DataFrame (do not mutate input)

    Args:
        df_iso: ISO-specific DataFrame

    Returns:
        Standardized DataFrame with canonical columns
    """
    # 1. Validate required columns exist
    required_cols = set(self.column_map.values())
    missing = required_cols - set(df_iso.columns)
    if missing:
        raise ValueError(f"Missing columns: {missing}")

    # 2. Create copy and rename columns (inverse mapping)
    inverse_map = {v: k for k, v in self.column_map.items()}
    df = df_iso.rename(columns=inverse_map).copy()

    # 3. Ensure datetime type
    df["interval_start_local"] = pd.to_datetime(df["interval_start_local"])

    # 4. Create gain_normalized
    df["gain_normalized"] = self.gain_normalizer(df)

    # 5. Add date column
    df["date"] = df["interval_start_local"].dt.date

    # 6. Add iso column
    df["iso"] = self.iso_name

    # 7. Run validations
    for validation in self.extra_validations:
        validation(df)

    return df
compute_capital_requirement 
compute_capital_requirement(df_std: DataFrame) -> float | None

Compute the capital requirement for the given standardized DataFrame.

Args: df_std: Standardized DataFrame (should be filtered to cleared trades)

Returns: Capital requirement value, or None if no capital_requirement_fn is defined

Source code in src/progridpy/metrics/schema.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
def compute_capital_requirement(self, df_std: pd.DataFrame) -> float | None:
    """
    Compute the capital requirement for the given standardized DataFrame.

    Args:
        df_std: Standardized DataFrame (should be filtered to cleared trades)

    Returns:
        Capital requirement value, or None if no capital_requirement_fn is defined
    """
    if self.capital_requirement_fn is None:
        return None
    return self.capital_requirement_fn(df_std)