Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def prepare_mediation_fee_config(
cli_token_to_flat_fee: Tuple[Tuple[TokenAddress, FeeAmount], ...],
cli_token_to_proportional_fee: Tuple[Tuple[TokenAddress, ProportionalFeeAmount], ...],
cli_token_to_proportional_imbalance_fee: Tuple[
Tuple[TokenAddress, ProportionalFeeAmount], ...
],
) -> MediationFeeConfig:
""" Converts the mediation fee CLI args to proper per-channel
mediation fees. """
tn_to_flat_fee: Dict[TokenAddress, FeeAmount] = {}
# Add the defaults for flat fees for DAI/WETH
tn_to_flat_fee[WETH_TOKEN_ADDRESS] = FeeAmount(DEFAULT_WETH_FLAT_FEE // 2)
tn_to_flat_fee[DAI_TOKEN_ADDRESS] = FeeAmount(DEFAULT_DAI_FLAT_FEE // 2)
# Store the flat fee settings for the given token addresses
# The given flat fee is for the whole mediation, but that includes two channels.
# Therefore divide by 2 here.
for address, fee in cli_token_to_flat_fee:
tn_to_flat_fee[address] = FeeAmount(fee // 2)
tn_to_proportional_fee: Dict[TokenAddress, ProportionalFeeAmount] = {
address: ppm_fee_per_channel(prop_fee)
for address, prop_fee in cli_token_to_proportional_fee
}
tn_to_proportional_imbalance_fee: Dict[TokenAddress, ProportionalFeeAmount] = {
address: prop_fee for address, prop_fee in cli_token_to_proportional_imbalance_fee
}
def test_update_fee(order, pathfinding_service_mock, token_network_model):
pathfinding_service_mock.database.insert(
"token_network", dict(address=token_network_model.address)
)
if order == "normal":
setup_channel(pathfinding_service_mock, token_network_model)
fee_schedule = FeeScheduleState(
flat=FeeAmount(1),
proportional=ProportionalFeeAmount(int(0.1e9)),
imbalance_penalty=[(TokenAmount(0), FeeAmount(0)), (TokenAmount(10), FeeAmount(10))],
)
fee_update = PFSFeeUpdate(
canonical_identifier=CanonicalIdentifier(
chain_identifier=ChainID(61),
token_network_address=token_network_model.address,
channel_identifier=ChannelID(1),
),
updating_participant=PARTICIPANT1,
fee_schedule=fee_schedule,
timestamp=datetime.utcnow(),
signature=EMPTY_SIGNATURE,
)
fee_update.sign(LocalSigner(PARTICIPANT1_PRIVKEY))
pathfinding_service_mock.handle_message(fee_update)
if order == "fee_update_before_channel_open":
def get_fee_update_message( # pylint: disable=too-many-arguments
updating_participant: Address,
chain_id=ChainID(61),
channel_identifier=DEFAULT_CHANNEL_ID,
token_network_address: TokenNetworkAddress = DEFAULT_TOKEN_NETWORK_ADDRESS,
fee_schedule: FeeScheduleState = FeeScheduleState(
cap_fees=True, flat=FeeAmount(1), proportional=ProportionalFeeAmount(1)
),
timestamp: datetime = datetime.utcnow(),
privkey_signer: bytes = PRIVATE_KEY_1,
) -> PFSFeeUpdate:
fee_message = PFSFeeUpdate(
canonical_identifier=CanonicalIdentifier(
chain_identifier=chain_id,
channel_identifier=channel_identifier,
token_network_address=token_network_address,
),
updating_participant=updating_participant,
fee_schedule=fee_schedule,
timestamp=timestamp,
signature=EMPTY_SIGNATURE,
)
]
)
# Make sure that routing works and the default fees are zero
assert tn.estimate_fee(1, 3) == 0
# Fees for the initiator are ignored
tn.set_fee(1, 2, flat=FA(1))
assert tn.estimate_fee(1, 3) == 0
# Node 2 demands fees for incoming transfers
tn.set_fee(2, 1, flat=FA(1))
assert tn.estimate_fee(1, 3) == 1
# Node 2 demands fees for outgoing transfers
tn.set_fee(2, 3, flat=FA(1))
assert tn.estimate_fee(1, 3) == 2
# No capacity in the opposite direction
assert tn.estimate_fee(3, 1) is None
# Reset fees to zero
tn.set_fee(1, 2)
tn.set_fee(2, 1)
tn.set_fee(2, 3)
# Let's try imbalance fees!
# When approximation iterations matter, those are given as sums of the steps.
# Incoming channel
# Without fee capping
tn.set_fee(2, 3, cap_fees=False)
def test_basic_fee():
flat_schedule = FeeScheduleState(flat=FA(2))
assert flat_schedule.fee(TA(10), capacity=TA(0)) == FA(2)
prop_schedule = FeeScheduleState(proportional=0.01)
assert prop_schedule.fee(TA(40), capacity=TA(0)) == FA(0)
assert prop_schedule.fee(TA(60), capacity=TA(0)) == FA(1)
assert prop_schedule.fee(TA(1000), capacity=TA(0)) == FA(10)
combined_schedule = FeeScheduleState(flat=FA(2), proportional=0.01)
assert combined_schedule.fee(TA(60), capacity=TA(0)) == FA(3)
def from_dict(cls, data: Dict[str, Any]) -> "TransferDescriptionWithSecretState":
restored = cls(
payment_network_identifier=to_canonical_address(data["payment_network_identifier"]),
payment_identifier=PaymentID(int(data["payment_identifier"])),
amount=PaymentAmount(int(data["amount"])),
allocated_fee=FeeAmount(int(data["allocated_fee"])),
token_network_identifier=to_canonical_address(data["token_network_identifier"]),
initiator=to_canonical_address(data["initiator"]),
target=to_canonical_address(data["target"]),
secret=deserialize_secret(data["secret"]),
)
return restored
amount=total,
view_out=view_out,
view_in=view_in,
amount_without_fees=total,
balance_in=view_in.capacity_partner,
balance_out=view_out.capacity,
schedule_in=view_in.fee_schedule_receiver,
schedule_out=view_out.fee_schedule_sender,
receivable_amount=view_in.capacity,
)
return None
fee = PaymentWithFeeAmount(amount_with_fees - total)
total += fee # type: ignore
fees.append(FeeAmount(fee))
# The hop to the target does not incur mediation fees
fees.append(FeeAmount(0))
return fees
def calculate_fee_margin(payment_amount: PaymentAmount, estimated_fee: FeeAmount) -> FeeAmount:
if estimated_fee == 0:
# If the total fees are zero, we assume that no fees are set. If the
# fees sum up to zero incidentally, we should add a margin, but we
# can't detect that case.
return FeeAmount(0)
return FeeAmount(
int(
ceil(
abs(estimated_fee) * DEFAULT_MEDIATION_FEE_MARGIN
+ payment_amount * PAYMENT_AMOUNT_BASED_FEE_MARGIN
)
def calculate_fee_margin(payment_amount: PaymentAmount, estimated_fee: FeeAmount) -> FeeAmount:
if estimated_fee == 0:
# If the total fees are zero, we assume that no fees are set. If the
# fees sum up to zero incidentally, we should add a margin, but we
# can't detect that case.
return FeeAmount(0)
return FeeAmount(
int(
ceil(
abs(estimated_fee) * DEFAULT_MEDIATION_FEE_MARGIN
+ payment_amount * PAYMENT_AMOUNT_BASED_FEE_MARGIN
)