"""
Implements the `DecentralizingAgent` which creates ony buy and one sell controller for each time-step and relinquishes
control of negotiations to buy/sell the required number of items of its input/output product.
"""
from typing import Tuple
import numpy as np
from negmas import LinearUtilityFunction
from scml.scml2020.components import (
IndependentNegotiationsManager,
MovingRangeNegotiationManager,
StepNegotiationManager,
SupplyDrivenProductionStrategy,
)
from ..components.prediction import MarketAwareTradePredictionStrategy
from ..components.signing import KeepOnlyGoodPrices
from ..components.trading import PredictionBasedTradingStrategy
from ..world import SCML2020Agent
__all__ = [
"DecentralizingAgent",
"IndDecentralizingAgent",
"DecentralizingAgentWithLogging",
"MarketAwareDecentralizingAgent",
"MarketAwareIndDecentralizingAgent",
]
class _NegotiationCallbacks:
def acceptable_unit_price(self, step: int, sell: bool) -> int:
production_cost = np.max(self.awi.profile.costs[:, self.awi.my_input_product])
if sell:
return min(
production_cost + self.input_cost[step:].min(), self.output_price[step]
)
return max(
self.output_price[step:].max() - production_cost, self.input_cost[step]
)
def target_quantity(self, step: int, sell: bool) -> int:
if sell:
needed, secured = (
self.outputs_needed[step:].sum(),
self.outputs_secured[step:].sum(),
)
else:
needed, secured = (
self.inputs_needed[:step].sum(),
self.inputs_secured[:step].sum(),
)
return min(self.awi.n_lines, needed - secured)
def target_quantities(self, steps: Tuple[int, int], sell: bool) -> np.ndarray:
"""Implemented for speed but not really required"""
if sell:
needed, secured = self.outputs_needed, self.outputs_secured
else:
needed, secured = self.inputs_needed, self.inputs_secured
return needed[steps[0] : steps[1]] - secured[steps[0] : steps[1]]
[docs]
class DecentralizingAgent(
_NegotiationCallbacks,
StepNegotiationManager,
PredictionBasedTradingStrategy,
SupplyDrivenProductionStrategy,
SCML2020Agent,
):
pass
[docs]
class MarketAwareDecentralizingAgent(
MarketAwareTradePredictionStrategy,
_NegotiationCallbacks,
MovingRangeNegotiationManager,
PredictionBasedTradingStrategy,
KeepOnlyGoodPrices,
SupplyDrivenProductionStrategy,
SCML2020Agent,
):
def __init__(
self,
*args,
buying_margin=None,
selling_margin=None,
min_price_margin=0.5,
max_price_margin=0.5,
**kwargs,
):
super().__init__(
*args,
buying_margin=buying_margin,
selling_margin=selling_margin,
min_price_margin=min_price_margin,
max_price_margin=max_price_margin,
**kwargs,
)
# def before_step(self):
# super().before_step()
# self.awi.loginfo_agent(
# f"Step {self.awi.current_step}\n{pformat(self.internal_state)}"
# )
[docs]
class DecentralizingAgentWithLogging(
_NegotiationCallbacks,
StepNegotiationManager,
PredictionBasedTradingStrategy,
SupplyDrivenProductionStrategy,
SCML2020Agent,
):
def __init__(self, *args, **kwargs):
super().__init__(*args, logdebug=True, **kwargs)
[docs]
class IndDecentralizingAgent(
_NegotiationCallbacks,
IndependentNegotiationsManager,
PredictionBasedTradingStrategy,
SupplyDrivenProductionStrategy,
SCML2020Agent,
):
[docs]
def create_ufun(self, is_seller: bool, issues=None, outcomes=None):
if is_seller:
return LinearUtilityFunction((1, 1, 10), issues=issues, outcomes=outcomes)
return LinearUtilityFunction((1, -1, -10), issues=issues, outcomes=outcomes)
[docs]
class MarketAwareIndDecentralizingAgent(
KeepOnlyGoodPrices,
MarketAwareTradePredictionStrategy,
IndDecentralizingAgent,
):
def __init__(
self,
*args,
buying_margin=None,
selling_margin=None,
min_price_margin=0.5,
max_price_margin=0.5,
**kwargs,
):
super().__init__(
*args,
buying_margin=buying_margin,
selling_margin=selling_margin,
min_price_margin=min_price_margin,
max_price_margin=max_price_margin,
**kwargs,
)