nexa-bidkit

Features

Installation

pip install nexa-bidkit

Quick Start

Constructing curves from DataFrames

import pandas as pd
from decimal import Decimal
from datetime import datetime
from zoneinfo import ZoneInfo
from nexa_bidkit import (
 CurveType,
 MTUDuration,
 MTUInterval,
 from_dataframe,
)
# Your bid data as a DataFrame
df = pd.DataFrame({
 "price": [10.5, 25.0, 45.0, 80.0],
 "volume": [50, 100, 75, 25],
})
# Define the MTU this curve applies to
mtu = MTUInterval.from_start(
 datetime(2026, 4, 1, 13, 0, tzinfo=ZoneInfo("Europe/Oslo")),
 MTUDuration.QUARTER_HOURLY
)
# Convert to a validated supply curve
curve = from_dataframe(df, curve_type=CurveType.SUPPLY, mtu=mtu)
print(f"Total volume: {curve.total_volume} MW")
print(f"Price range: {curve.min_price} - {curve.max_price} EUR/MWh")

Combining multiple bids

from nexa_bidkit import merge_curves
# Combine bids from multiple plants
plant_curves = [
 from_dataframe(plant1_df, curve_type=CurveType.SUPPLY, mtu=mtu),
 from_dataframe(plant2_df, curve_type=CurveType.SUPPLY, mtu=mtu),
]
portfolio = merge_curves(plant_curves, aggregation="sum")

Applying constraints

from nexa_bidkit import clip_curve, filter_zero_volume, aggregate_by_price
# Clean and constrain curve
curve = from_dataframe(df, curve_type=CurveType.SUPPLY, mtu=mtu)
curve = filter_zero_volume(curve) # Remove empty steps
curve = clip_curve(
 curve,
 min_price=Decimal("0"), # No negative prices
 max_volume=Decimal("500") # Capacity constraint
)
curve = aggregate_by_price(curve) # Consolidate for smaller message size

Creating bids

from decimal import Decimal
from datetime import datetime
from zoneinfo import ZoneInfo
from nexa_bidkit import (
 BiddingZone,
 Direction,
 DeliveryPeriod,
 MTUDuration,
 block_bid,
 indivisible_block_bid,
 linked_block_bid,
 exclusive_group,
)
# Define a delivery period (4 hours of quarter-hourly MTUs)
delivery = DeliveryPeriod(
 start=datetime(2026, 4, 1, 10, 0, tzinfo=ZoneInfo("Europe/Oslo")),
 end=datetime(2026, 4, 1, 14, 0, tzinfo=ZoneInfo("Europe/Oslo")),
 duration=MTUDuration.QUARTER_HOURLY,
)
# Create a block bid (partially fillable)
peak_bid = block_bid(
 bidding_zone=BiddingZone.NO1,
 direction=Direction.SELL,
 delivery_period=delivery,
 price=Decimal("55.0"),
 volume=Decimal("100"),
 min_acceptance_ratio=Decimal("0.5"), # Accept 50%+ fill
)
# Create an indivisible block bid (all-or-nothing)
must_run = indivisible_block_bid(
 bidding_zone=BiddingZone.NO1,
 direction=Direction.SELL,
 delivery_period=delivery,
 price=Decimal("25.0"),
 volume=Decimal("50"),
)
# Create a linked block bid (only accepted if parent accepted)
ramp_up = linked_block_bid(
 parent_bid_id=must_run.bid_id,
 bidding_zone=BiddingZone.NO1,
 direction=Direction.SELL,
 delivery_period=delivery,
 price=Decimal("35.0"),
 volume=Decimal("25"),
)
# Create an exclusive group (at most one accepted)
option_a = block_bid(
 bidding_zone=BiddingZone.NO1,
 direction=Direction.SELL,
 delivery_period=delivery,
 price=Decimal("40.0"),
 volume=Decimal("150"),
)
option_b = block_bid(
 bidding_zone=BiddingZone.NO1,
 direction=Direction.SELL,
 delivery_period=delivery,
 price=Decimal("45.0"),
 volume=Decimal("120"),
)
options = exclusive_group([option_a, option_b])
print(f"Peak bid total volume: {peak_bid.total_volume} MW")
print(f"Must-run is indivisible: {must_run.is_indivisible}")
print(f"Exclusive group has {options.member_count} options")

Managing portfolios with OrderBook

from nexa_bidkit import (
 create_order_book,
 add_bid,
 add_bids,
 get_bids_by_zone,
 get_bids_by_status,
 count_bids,
 total_volume_by_zone,
 update_all_statuses,
 orders_to_dataframe,
 BidStatus,
)
# Create an order book
book = create_order_book()
# Add individual bids
book = add_bid(book, must_run)
book = add_bid(book, peak_bid)
# Add multiple bids at once
book = add_bids(book, [ramp_up, options])
# Query bids
no1_bids = get_bids_by_zone(book, BiddingZone.NO1)
draft_bids = get_bids_by_status(book, BidStatus.DRAFT)
# Aggregate statistics
bid_counts = count_bids(book)
volumes = total_volume_by_zone(book)
print(f"Total bids: {sum(bid_counts.values())}")
print(f"NO1 volume: {volumes[BiddingZone.NO1]} MW")
# Update statuses (e.g., after validation)
book = update_all_statuses(book, BidStatus.VALIDATED)
# Export to pandas for analysis
df = orders_to_dataframe(book)
print(df[["bid_id", "bid_type", "bidding_zone", "status"]])

Validating bids for EUPHEMIA compliance

from nexa_bidkit import (
 validate_bid,
 validate_bids,
 validate_order_book_for_submission,
 get_validation_summary,
 EuphemiaValidationError,
 DataQualityError,
 TemporalValidationError,
)
# Validate individual bid
try:
 validate_bid(peak_bid)
 print("Bid is valid!")
except EuphemiaValidationError as e:
 print(f"EUPHEMIA compliance error: {e}")
except DataQualityError as e:
 print(f"Data quality issue: {e}")
# Batch validation (collects all errors)
results = validate_bids([must_run, peak_bid, ramp_up])
summary = get_validation_summary(results)
print(f"Validated {summary['total_bids']} bids")
print(f"Pass rate: {summary['pass_rate']:.1f}%")
print(f"Errors by type: {summary['error_types']}")
# Comprehensive validation before submission
gate_closure = datetime(2026, 3, 31, 12, 0, tzinfo=ZoneInfo("Europe/Oslo"))
try:
 validate_order_book_for_submission(
 book,
 gate_closure_time=gate_closure,
 )
 print("Order book ready for submission")
except ValidationError as e:
 print(f"Validation failed: {e}")

Submitting to Nord Pool

Curve order from a SimpleBid

from nexa_bidkit.nordpool import simple_bid_to_curve_order
from nexa_bidkit import (
 BiddingZone, CurveType, Direction, MTUDuration, MTUInterval,
 PriceQuantityCurve, PriceQuantityStep, SimpleBid,
)
from decimal import Decimal
from datetime import datetime
from zoneinfo import ZoneInfo
mtu = MTUInterval.from_start(
 datetime(2026, 4, 1, 13, 0, tzinfo=ZoneInfo("Europe/Oslo")),
 MTUDuration.HOURLY,
)
curve = PriceQuantityCurve(
 curve_type=CurveType.SUPPLY,
 steps=[
 PriceQuantityStep(price=Decimal("10.00"), volume=Decimal("50")),
 PriceQuantityStep(price=Decimal("20.00"), volume=Decimal("100")),
 ],
 mtu=mtu,
)
bid = SimpleBid(
 bid_id="simple-1",
 bidding_zone=BiddingZone.NO1,
 direction=Direction.SELL,
 curve=curve,
)
# Your resolver maps (MTUInterval, BiddingZone) → Nord Pool contract ID.
# Call Nord Pool's products API to populate this lookup at runtime.
def resolve_contract(mtu, zone):
 hour = mtu.start.hour
 return f"{zone.value}-{hour}"
payload = simple_bid_to_curve_order(
 bid,
 auction_id="DA-2026年04月01日",
 portfolio="my-portfolio",
 contract_id_resolver=resolve_contract,
)
# Serialise to JSON for the Nord Pool API (uses camelCase aliases)
print(payload.model_dump(by_alias=True))
# {
# "auctionId": "DA-2026年04月01日",
# "portfolio": "my-portfolio",
# "areaCode": "NO1",
# "comment": null,
# "curves": [{"contractId": "NO1-13", "curvePoints": [...]}]
# }

Block and linked block orders

from nexa_bidkit.nordpool import block_bid_to_block_list, linked_block_bid_to_block_list
from nexa_bidkit import (
 BiddingZone, DeliveryPeriod, Direction, MTUDuration,
 block_bid, linked_block_bid,
)
from decimal import Decimal
from datetime import datetime
from zoneinfo import ZoneInfo
tz = ZoneInfo("Europe/Oslo")
delivery = DeliveryPeriod(
 start=datetime(2026, 4, 1, 10, 0, tzinfo=tz),
 end=datetime(2026, 4, 1, 14, 0, tzinfo=tz),
 duration=MTUDuration.HOURLY,
)
must_run = block_bid(
 bidding_zone=BiddingZone.NO1,
 direction=Direction.SELL,
 delivery_period=delivery,
 price=Decimal("25.0"),
 volume=Decimal("50"),
 bid_id="must-run",
)
ramp_up = linked_block_bid(
 parent_bid_id=must_run.bid_id,
 bidding_zone=BiddingZone.NO1,
 direction=Direction.SELL,
 delivery_period=delivery,
 price=Decimal("35.0"),
 volume=Decimal("25"),
)
block_payload = block_bid_to_block_list(must_run, "DA-2026年04月01日", "my-portfolio", resolve_contract)
linked_payload = linked_block_bid_to_block_list(ramp_up, "DA-2026年04月01日", "my-portfolio", resolve_contract)
# The linked block payload carries the parent reference
print(linked_payload.blocks[0].linked_to) # "must-run"

Converting a whole OrderBook at once

from nexa_bidkit.nordpool import order_book_to_nord_pool
from nexa_bidkit import create_order_book, add_bids
book = create_order_book()
book = add_bids(book, [must_run, ramp_up])
submission = order_book_to_nord_pool(
 book,
 auction_id="DA-2026年04月01日",
 portfolio="my-portfolio",
 contract_id_resolver=resolve_contract,
)
print(f"Curve orders: {len(submission.curve_orders)}")
print(f"Block orders: {len(submission.block_orders)}")
print(f"Linked block orders: {len(submission.linked_block_orders)}")
print(f"Exclusive groups: {len(submission.exclusive_group_orders)}")

Submitting to EXAA

Hourly and 15-minute bids from an OrderBook

from nexa_bidkit.exaa import (
 order_book_to_exaa,
 standard_hourly_product_id,
 ExaaOrderType,
)
from nexa_bidkit import create_order_book, add_bids
book = create_order_book()
book = add_bids(book, [must_run, peak_bid])
# Use standard product ID helpers, or supply your own resolver
# from the auction's products API response.
payload = order_book_to_exaa(
 book,
 account_id="APTAP1",
 product_id_resolver=standard_hourly_product_id,
 order_type=ExaaOrderType.STEP,
)
# Serialise to JSON for the EXAA API (uses camelCase aliases)
print(payload.model_dump(by_alias=True))
# {
# "units": {"price": "EUR", "volume": "MWh/h"},
# "orders": [{
# "accountID": "APTAP1",
# "hourlyProducts": {"typeOfOrder": "STEP", "products": [...]},
# ...
# }]
# }

Block bids

from nexa_bidkit.exaa import order_book_to_exaa, standard_hourly_product_id
def my_block_resolver(period):
 # Map delivery periods to EXAA block product IDs from the auction response
 return "bEXAbase (01-24)"
payload = order_book_to_exaa(
 book,
 account_id="APTAP1",
 product_id_resolver=standard_hourly_product_id,
 block_product_resolver=my_block_resolver,
)

Examples

Running the examples

# Install with dev dependencies (includes jupyter, matplotlib)
poetry install
# Run a notebook interactively
poetry run jupyter notebook examples/01_simple_hourly_bids.ipynb
# Execute all notebooks and update outputs in-place
make execute-notebooks
# Run notebooks as tests (used by CI)
make test-notebooks

Core Concepts

Market Time Units (MTU)

Price-Quantity Curves

EUPHEMIA Bid Types

Development

Setup

# Install poetry
curl -sSL https://install.python-poetry.org | python3 -
# Install dependencies
poetry install
# Run tests
poetry run pytest
# Type checking
poetry run mypy src
# Linting
poetry run ruff check src

Running tests

# All tests with coverage
make test
# Or using poetry directly
poetry run pytest tests/ -v
poetry run pytest --cov=nexa_bidkit --cov-report=term-missing

Releasing a new version

Prerequisites (one-time, per environment)

Publishing a beta release

# 1. Create a version bump branch (main is protected — never push directly)
git checkout main && git pull
git checkout -b chore/bump-version-to-1.0.0b1
# 2. Bump the version in pyproject.toml
make bump version=1.0.0b1
# 3. Commit, push, and open a PR
git add pyproject.toml
git commit -m "chore: bump version to 1.0.0b1"
git push -u origin chore/bump-version-to-1.0.0b1
gh pr create --title "chore: bump version to 1.0.0b1" --body "Version bump for beta release."
# 4. Merge the PR into main, then on GitHub create a new Release
# - Tag: v1.0.0b1 (target: main)
# - Check "This is a pre-release"
# - Publish release
# - Delete the chore branches

Publishing a stable release

# 1. Create a version bump branch (main is protected — never push directly)
git checkout main && git pull
git checkout -b chore/bump-version-to-1.0.0
# 2. Bump the version in pyproject.toml
make bump version=1.0.0
# 3. Commit, push, and open a PR
git add pyproject.toml
git commit -m "chore: bump version to 1.0.0"
git push -u origin chore/bump-version-to-1.0.0
gh pr create --title "chore: bump version to 1.0.0" --body "Version bump for stable release."
# 4. Merge the PR into main, then on GitHub create a new Release
# - Tag: v1.0.0 (target: main)
# - Do NOT check "This is a pre-release"
# - Publish release
# - Delete the chore branches

What happens next

validate → ci → build → publish-testpypi → [approval] → publish-pypi

Verify the release

# Check test.pypi.org first (no approval needed)
pip install --index-url https://test.pypi.org/simple/ nexa-bidkit==1.0.0b1
# After approving the pypi gate, check the stable index
pip install nexa-bidkit==1.0.0
# Pre-releases require --pre
pip install --pre nexa-bidkit

Local pre-flight check

make publish-check tag=v1.0.0
# OK: 1.0.0 matches pyproject.toml

License

Contributing