amt.py: new UniAmt type, equivalent to BTCAmt

This commit is contained in:
The MMGen Project 2025-03-17 19:16:46 +03:00
commit eb80abaaee
Signed by: mmgen
GPG key ID: 3F8B1861E32B7DA2
6 changed files with 20 additions and 12 deletions

View file

@ -173,6 +173,9 @@ class BTCAmt(CoinAmt):
atomic = satoshi
units = ('satoshi',)
class UniAmt(BTCAmt):
coin = None
class BCHAmt(BTCAmt):
coin = 'BCH'

View file

@ -13,6 +13,7 @@ swap.proto.thorchain.memo: THORChain swap protocol memo class
"""
from ....util import die
from ....amt import UniAmt
from . import name as proto_name
@ -125,9 +126,9 @@ class Memo:
self.proto = proto
self.chain = chain or proto.coin
if trade_limit is None:
self.trade_limit = self.proto.coin_amt('0')
self.trade_limit = UniAmt('0')
else:
assert type(trade_limit) is self.proto.coin_amt, f'{type(trade_limit)} != {self.proto.coin_amt}'
assert type(trade_limit) is UniAmt, f'{type(trade_limit)} != {UniAmt}'
self.trade_limit = trade_limit
from ....addr import is_coin_addr
assert is_coin_addr(proto, addr)

View file

@ -13,6 +13,7 @@ swap.proto.thorchain.thornode: THORChain swap protocol network query ops
"""
import json
from ....amt import UniAmt
class ThornodeRPCClient:
@ -73,12 +74,12 @@ class Thornode:
in_coin = tx.send_proto.coin
out_coin = tx.recv_proto.coin
in_amt = self.in_amt
out_amt = tx.recv_proto.coin_amt(int(d['expected_amount_out']), from_unit='satoshi')
out_amt = UniAmt(int(d['expected_amount_out']), from_unit='satoshi')
if trade_limit:
from . import ExpInt4
e = ExpInt4(trade_limit.to_unit('satoshi'))
tl_rounded = tx.recv_proto.coin_amt(e.trunc, from_unit='satoshi')
tl_rounded = UniAmt(e.trunc, from_unit='satoshi')
ratio = usr_trade_limit if type(usr_trade_limit) is float else float(tl_rounded / out_amt)
direction = 'ABOVE' if ratio > 1 else 'below'
mcolor, lblcolor = (
@ -102,13 +103,13 @@ class Thornode:
else:
ymsg('Warning: unknown gas unit ‘{}’, cannot estimate fee'.format(d['gas_rate_units']))
min_in_amt = tx.send_proto.coin_amt(int(d['recommended_min_amount_in']), from_unit='satoshi')
min_in_amt = UniAmt(int(d['recommended_min_amount_in']), from_unit='satoshi')
gas_unit = {
'satsperbyte': 'sat/byte',
}.get(d['gas_rate_units'], d['gas_rate_units'])
elapsed_disp = format_elapsed_hr(d['expiry'], future_msg='from now')
fees = d['fees']
fees_t = tx.recv_proto.coin_amt(int(fees['total']), from_unit='satoshi')
fees_t = UniAmt(int(fees['total']), from_unit='satoshi')
fees_pct_disp = str(fees['total_bps'] / 100) + '%'
slip_pct_disp = str(fees['slippage_bps'] / 100) + '%'
hdr = f'SWAP QUOTE (source: {self.rpc.host})'

View file

@ -13,6 +13,7 @@ tx.new_swap: new swap transaction class
"""
from .new import New
from ..amt import UniAmt
class NewSwap(New):
desc = 'swap transaction'
@ -28,7 +29,7 @@ class NewSwap(New):
if s := self.cfg.trade_limit:
self.usr_trade_limit = (
1 - float(s[:-1]) / 100 if s.endswith('%') else
self.recv_proto.coin_amt(self.cfg.trade_limit))
UniAmt(self.cfg.trade_limit))
else:
self.usr_trade_limit = None
@ -40,11 +41,11 @@ class NewSwap(New):
from ..term import get_char
def get_trade_limit():
if type(self.usr_trade_limit) is self.recv_proto.coin_amt:
if type(self.usr_trade_limit) is UniAmt:
return self.usr_trade_limit
elif type(self.usr_trade_limit) is float:
return (
self.recv_proto.coin_amt(int(c.data['expected_amount_out']), from_unit='satoshi')
UniAmt(int(c.data['expected_amount_out']), from_unit='satoshi')
* self.usr_trade_limit)
while True:

View file

@ -15,6 +15,7 @@ test.cmdtest_d.httpd.thornode: Thornode WSGI http server
import time, re, json
from mmgen.cfg import Config
from mmgen.amt import UniAmt
from . import HTTPD
@ -75,7 +76,7 @@ class ThornodeServer(HTTPD):
from mmgen.protocol import init_proto
send_proto = init_proto(cfg, send_coin, network='regtest', need_amt=True)
in_amt = send_proto.coin_amt(int(amt_atomic), from_unit='satoshi')
in_amt = UniAmt(int(amt_atomic), from_unit='satoshi')
out_amt = in_amt * (prices[send_coin] / prices[recv_coin])
addr = make_inbound_addr(send_proto, send_proto.preferred_mmtypes[0])

View file

@ -170,6 +170,7 @@ class unit_tests:
def memo(self, name, ut, desc='Swap transaction memo'):
from mmgen.protocol import init_proto
from mmgen.amt import UniAmt
from mmgen.swap.proto.thorchain.memo import Memo
for coin, addrtype in (
('ltc', 'bech32'),
@ -186,13 +187,13 @@ class unit_tests:
(None, 0),
):
vmsg('\nTesting memo initialization:')
m = Memo(proto, addr, trade_limit=proto.coin_amt(limit) if limit else None)
m = Memo(proto, addr, trade_limit=UniAmt(limit) if limit else None)
vmsg(f'str(memo): {m}')
vmsg(f'repr(memo): {m!r}')
vmsg(f'limit: {limit}')
p = Memo.parse(m)
limit_dec = proto.coin_amt(p.trade_limit, from_unit='satoshi')
limit_dec = UniAmt(p.trade_limit, from_unit='satoshi')
vmsg(f'limit_dec: {limit_dec.hl()}')
vmsg('\nTesting memo parsing:')