Skip to content

market

cognite.powerops.resync.config.Market

Bases: Configuration

Source code in cognite/powerops/resync/config/market/market.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
class Market(Configuration):
    parent_external_id: ClassVar[str] = "market_configurations"
    external_id: str
    name: str
    max_price: Optional[float] = None
    min_price: Optional[float] = None
    time_unit: Optional[str] = None
    timezone: Optional[str] = None
    tick_size: Optional[float] = None
    trade_lot: Optional[float] = None
    price_steps: Optional[int] = None
    price_unit: Optional[str] = None

Dayahead

cognite.powerops.resync.config.BidProcessConfig

Bases: Configuration

Source code in cognite/powerops/resync/config/market/dayahead.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class BidProcessConfig(Configuration):
    name: str
    price_area_name: str = Field(alias="bid_price_area")
    price_scenarios: list[PriceScenarioID] = Field(alias="bid_price_scenarios")
    main_scenario: str = Field(alias="bid_main_scenario")
    bid_date: Optional[RelativeTime] = None
    shop_start: Optional[RelativeTime] = Field(None, alias="shop_starttime")
    shop_end: Optional[RelativeTime] = Field(None, alias="shop_endtime")
    bid_matrix_generator: str = Field(alias="bid_bid_matrix_generator_config_external_id")
    price_scenarios_per_watercourse: Optional[dict[str, set[str]]] = None
    is_default_config_for_price_area: bool = False
    no_shop: bool = Field(False, alias="no_shop")

    @field_validator("shop_start", "shop_end", "bid_date", mode="before")
    @classmethod
    def json_loads(cls, value):
        return {"operations": json.loads(value)} if isinstance(value, str) else value

    @field_validator("price_scenarios", mode="after")
    @classmethod
    def ensure_no_duplicates(cls, value: list[PriceScenarioID], info: FieldValidationInfo):
        scenario_ids = Counter(scenario.id for scenario in value)
        if duplicates := [scenario_id for scenario_id, count in scenario_ids.items() if count > 1]:
            bidprocess_name = info.data.get("name", "unknown")
            raise ValueError(
                f"Duplicate price scenarios for bidprocess {bidprocess_name} was found: {list(duplicates)}"
            )
        return value

cognite.powerops.resync.config.BidMatrixGeneratorConfig

Bases: BaseModel

Source code in cognite/powerops/resync/config/market/dayahead.py 
14
15
16
17
18
19
20
21
22
class BidMatrixGeneratorConfig(BaseModel):
    name: str
    default_method: str
    default_function_external_id: str
    column_external_ids: ClassVar[list[str]] = [
        "shop_plant",
        "bid_matrix_generation_method",
        "function_external_id",
    ]

RKOM

cognite.powerops.resync.config.RkomMarketConfig

Bases: BaseModel

Source code in cognite/powerops/resync/config/market/rkom.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
class RkomMarketConfig(BaseModel):
    external_id: str
    name: str
    timezone: str
    start_of_week: int

    @classmethod
    def default(cls) -> RkomMarketConfig:
        return cls(
            name="RKOM weekly (Statnett)",
            timezone="Europe/Oslo",
            start_of_week=1,
            external_id="market_configuration_statnett_rkom_weekly",
        )

cognite.powerops.resync.config.RKOMBidProcessConfig

Bases: Configuration

Source code in cognite/powerops/resync/config/market/rkom.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
class RKOMBidProcessConfig(Configuration):
    watercourse: str = Field(alias="bid_watercourse")

    price_scenarios: list[PriceScenarioID] = Field(alias="bid_price_scenarios")
    reserve_scenarios: ReserveScenarios = Field(alias="bid_reserve_scenarios")

    shop_start: RelativeTime = Field(alias="shop_starttime")
    shop_end: RelativeTime = Field(alias="shop_endtime")

    timezone: str = "Europe/Oslo"
    method: str = "simple"

    minimum_price: int = 0  # TODO: need to specify currency
    price_premium: int = 0  # TODO: need to specify currency

    parent_external_id: typing.ClassVar[str] = "rkom_bid_process_configurations"
    mapping_type: ClassVar[str] = "rkom_incremental_mapping"

    @model_validator(mode="before")
    @classmethod
    def create_reserve_scenarios(cls, value):
        if not isinstance(volumes := value.get("reserve_scenarios"), str):
            return value
        volumes = [int(volume.removesuffix("MW")) for volume in volumes[1:-1].split(",")]

        value["bid_reserve_scenarios"] = dict(
            volumes=volumes,
            auction=value["bid_auction"],
            product=value["bid_product"],
            block=value["bid_block"],
            reserve_group=value["labels"][0]["externalId"],
            mip_plant_time_series=[],
        )
        return value

    @field_validator("shop_start", "shop_end", mode="before")
    @classmethod
    def json_loads(cls, value):
        return {"operations": json.loads(value)} if isinstance(value, str) else value

    @field_validator("price_scenarios", mode="before")
    @classmethod
    def literal_eval(cls, value):
        return [{"id": id_} for id_ in ast.literal_eval(value)] if isinstance(value, str) else value

    @property
    def sorted_volumes(self) -> list[int]:
        return sorted(self.reserve_scenarios.volumes)

    @property
    def name(self) -> str:
        return (
            f"{self.watercourse}_"
            f"{self.reserve_scenarios.auction.value}_"
            f"{self.reserve_scenarios.product}_"
            f"{self.reserve_scenarios.block}_"
            f"{len(self.price_scenarios)}-prices_"
            f"{self.sorted_volumes[1]}MW-{self.sorted_volumes[-1]}MW"
        )

    @property
    def external_id(self) -> str:
        return f"POWEROPS_{self.name}"

    @property
    def bid_date(self) -> RelativeTime:
        if self.reserve_scenarios.auction == "week":
            return RelativeTime(relative_time_string="monday")
        else:
            return RelativeTime(relative_time_string="saturday")

    @property
    def rkom_plants(self) -> list[str]:
        return [plant for plant, _ in self.reserve_scenarios.mip_plant_time_series]

cognite.powerops.resync.config.RKOMBidCombinationConfig

Bases: Configuration

Source code in cognite/powerops/resync/config/market/rkom.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
class RKOMBidCombinationConfig(Configuration):
    parent_external_id: ClassVar[str] = "rkom_bid_combination_configurations"
    model_config: ClassVar[ConfigDict] = ConfigDict(populate_by_name=True)
    auction: Auction = Field(alias="bid_auction")
    name: str = Field("default", alias="bid_combination_name")
    rkom_bid_config_external_ids: list[str] = Field(alias="bid_rkom_bid_configs")

    @field_validator("auction", mode="before")
    @classmethod
    def to_enum(cls, value):
        return Auction[value] if isinstance(value, str) else value

    @field_validator("rkom_bid_config_external_ids", mode="before")
    @classmethod
    def parse_string(cls, value):
        return [external_id for external_id in ast.literal_eval(value)] if isinstance(value, str) else value

cognite.powerops.resync.config.ReserveScenario dataclass

Source code in cognite/powerops/resync/config/market/rkom.py 
 33
 34
 35
 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
 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
@dataclass
class ReserveScenario:
    volume: int
    auction: Auction
    product: Product
    block: Block
    reserve_group: str
    mip_plant_time_series: list[
        tuple[PlantExternalId, Optional[TimeSeriesExternalId]]
    ]  # (plant, mip_flag_time_series) for plants with generators that are in the reserve group
    obligation_external_id: Optional[str] = None

    def __post_init__(self):
        if len(self.mip_plant_time_series) == 0:
            raise ValueError("No `mip_plants` specified!")

    @property
    def obligation_transformations(self) -> list[Transformation]:
        # TODO: move some of this logic
        n_days = 5 if self.auction == "week" else 2
        night = self.block == "night"
        kwargs = _generate_reserve_schedule(volume=self.volume, n_days=n_days, night=night)
        if self.obligation_external_id:
            return [Transformation(transformation=TransformationType.DYNAMIC_ADD_FROM_OFFSET, kwargs=kwargs)]
        else:
            return [Transformation(transformation=TransformationType.DYNAMIC_STATIC, kwargs=kwargs)]

    def mip_flag_transformations(self, mip_time_series: Optional[str]) -> list[Transformation]:
        if not mip_time_series:
            # Just run with_mip flag the entire period
            return [Transformation(transformation=TransformationType.STATIC, kwargs={"0": 1})]

        # Run with mip_flag during bid, in addition to what is already set
        n_days = 5 if self.auction == "week" else 2
        n_minutes = n_days * 24 * 60
        kwargs = {0: 1, n_minutes: 0}
        return [
            Transformation(transformation=TransformationType.DYNAMIC_ADD_FROM_OFFSET, kwargs=kwargs),
            Transformation(transformation=TransformationType.TO_BOOL),
        ]

    @property
    def obligation_object_type(self) -> str:
        return "reserve_group"

    @property
    def obligation_attribute_name(self) -> str:
        if self.product == "up":
            return "rr_up_obligation"
        elif self.product == "down":
            return "rr_down_obligation"

    def to_time_series_mapping(self) -> TimeSeriesMapping:
        obligation = TimeSeriesMappingEntry(
            object_type=self.obligation_object_type,
            object_name=self.reserve_group,
            attribute_name=self.obligation_attribute_name,
            time_series_external_id=self.obligation_external_id,
            transformations=self.obligation_transformations,
            retrieve=RetrievalType.RANGE if self.obligation_external_id else None,
        )

        mip_flags = [
            TimeSeriesMappingEntry(
                object_type="plant",
                object_name=plant_name,
                attribute_name="mip_flag",
                time_series_external_id=mip_time_series,
                transformations=self.mip_flag_transformations(mip_time_series),
                retrieve=RetrievalType.RANGE if mip_time_series else None,
                aggregation=AggregationMethod.max,  # TODO: or `AggregationMethod.first`?
            )
            for plant_name, mip_time_series in self.mip_plant_time_series
        ]

        return TimeSeriesMapping(rows=[obligation, *mip_flags])

cognite.powerops.resync.config.ReserveScenarios

Bases: BaseModel

Source code in cognite/powerops/resync/config/market/rkom.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
class ReserveScenarios(BaseModel):
    volumes: list[int]
    auction: Auction
    product: Product
    block: Block
    reserve_group: str
    mip_plant_time_series: list[tuple[PlantExternalId, Optional[TimeSeriesExternalId]]]
    obligation_external_id: Optional[str]

    @field_validator("auction", mode="before")
    @classmethod
    def to_enum(cls, value):
        return Auction[value] if isinstance(value, str) else value

    @field_validator("volumes", mode="before")
    @classmethod
    def valid_volumes(cls, volumes):
        if 0 not in volumes:
            raise ValueError("You probably want 0 MW as one of the volumes!")
        if any(volume < 0 for volume in volumes):
            raise ValueError(f"All volumes should be positive! Got {volumes}")
        return list(set(volumes))  # Do not want duplicate volumes

    def __str__(self) -> str:
        return json.dumps([f"{volume}MW" for volume in sorted(self.volumes)])

    def __len__(self) -> int:
        return len(self.list_scenarios())

    def __iter__(self) -> Generator[ReserveScenario, None, None]:
        yield from self.list_scenarios()

    def list_scenarios(self) -> list[ReserveScenario]:
        return [
            ReserveScenario(
                volume=volume,
                auction=self.auction,
                product=self.product,
                block=self.block,
                reserve_group=self.reserve_group,
                mip_plant_time_series=self.mip_plant_time_series,
                obligation_external_id=self.obligation_external_id,
            )
            for volume in self.volumes
        ]

cognite.powerops.resync.config.Product: TypeAlias = Literal['up', 'down'] module-attribute

cognite.powerops.resync.config.Block: TypeAlias = Literal['day', 'night'] module-attribute

Benchmarking

cognite.powerops.resync.config.BenchmarkingConfig

Bases: Configuration

Source code in cognite/powerops/resync/config/market/benchmarking.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class BenchmarkingConfig(Configuration):
    model_config = ConfigDict(populate_by_name=True)
    bid_date: RelativeTime
    shop_start: RelativeTime = Field(alias="shop_starttime")
    shop_end: RelativeTime = Field(alias="shop_endtime")
    production_plan_time_series: Optional[dict[str, list[str]]] = Field(
        default_factory=dict, alias="bid_production_plan_time_series"
    )
    market_config_external_id: str = Field(alias="bid_market_config_external_id")
    bid_process_configuration_assets: list[str] = []  # noqa: RUF012
    relevant_shop_objective_metrics: dict[str, str] = {  # noqa: RUF012
        "grand_total": "Grand Total",
        "total": "Total",
        "sum_penalties": "Sum Penalties",
        "major_penalties": "Major Penalties",
        "minor_penalties": "Minor Penalties",
        "load_value": "Load Value",
        "market_sale_buy": "Market Sale Buy",
        "rsv_end_value_relative": "RSV End Value Relative",
        "startup_costs": "Startup Costs",
        "vow_in_transit": "Vow in Transit",
        "sum_feeding_fee": "Sum Feeding Fee",
        "rsv_tactical_penalty": "RSV Tactical Penalty",
        "rsv_end_value": "RSV End Value",
        "bypass_cost": "Bypass Cost",
        "gate_discharge_cost": "Gate Discharge Cost",
        "reserve_violation_penalty": "Reserve Violation Penalty",
        "load_penalty": "Load Penalty",
    }  # Pydantic handles mutable defaults such that this is OK:
    # https://stackoverflow.com/questions/63793662/how-to-give-a-pydantic-list-field-a-default-value/63808835#63808835

    # TODO: Consider adding relationships to bid process config
    #  assets (or remove the optional part that uses those relationships in power-ops-functions)

    @field_validator("shop_start", "shop_end", "bid_date", mode="before")
    @classmethod
    def json_loads(cls, value):
        return {"operations": json.loads(value)} if isinstance(value, str) else value

Shared

cognite.powerops.resync.config.PriceScenario

Bases: BaseModel

Source code in cognite/powerops/resync/config/market/_core.py 
 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
class PriceScenario(BaseModel):
    name: str
    time_series_external_id: Optional[str] = None
    transformations: Optional[list[Transformation]] = None

    def to_time_series_mapping(self) -> TimeSeriesMapping:
        retrieve = RetrievalType.RANGE if self.time_series_external_id else None
        transformations = self.transformations or []

        # to make buy price slightly higher than sale price in SHOP
        transformations_buy_price = [
            *transformations,
            Transformation(transformation=TransformationType.ADD, kwargs={"value": 0.01}),
        ]

        sale_price_row = TimeSeriesMappingEntry(
            object_type="market",
            object_name=self.name,
            attribute_name="sale_price",
            time_series_external_id=self.time_series_external_id,
            transformations=transformations,
            retrieve=retrieve,
            aggregation=AggregationMethod.mean,
        )

        buy_price_row = TimeSeriesMappingEntry(
            object_type="market",
            object_name=self.name,
            attribute_name="buy_price",
            time_series_external_id=self.time_series_external_id,
            transformations=transformations_buy_price,
            retrieve=retrieve,
            aggregation=AggregationMethod.mean,
        )

        return TimeSeriesMapping(rows=[sale_price_row, buy_price_row])

cognite.powerops.resync.config.PriceScenarioID

Bases: BaseModel

Source code in cognite/powerops/resync/config/market/_core.py 
25
26
27
class PriceScenarioID(BaseModel):
    id: str
    rename: str = ""