_('Unconfirmed'),

_('Unconfirmed parent'),

_('Not Verified'),

_('Local'),

if txin_type in ('p2pkh', 'p2wpkh', 'p2wpkh-p2sh'):

address = bitcoin.pubkey_to_address(txin_type, pubkey)

scripthash = bitcoin.address_to_scripthash(address)

elif txin_type == 'p2pk':

script = bitcoin.public_key_to_p2pk_script(pubkey)

scripthash = bitcoin.script_to_scripthash(script)

address = None

else:

raise Exception(f'unexpected txin_type to sweep: {txin_type}')

u = network.run_from_another_thread(network.listunspent_for_scripthash(scripthash))

for item in u:

if len(inputs) >= imax:

break

prevout_str = item['tx_hash'] + ':%d' % item['tx_pos']

prevout = TxOutpoint.from_str(prevout_str)

utxo = PartialTxInput(prevout=prevout)

utxo._trusted_value_sats = int(item['value'])

utxo._trusted_address = address

utxo.block_height = int(item['height'])

utxo.script_type = txin_type

utxo.pubkeys = [bfh(pubkey)]

utxo.num_sig = 1

if txin_type == 'p2wpkh-p2sh':

utxo.redeem_script = bfh(bitcoin.p2wpkh_nested_script(pubkey))

def find_utxos_for_privkey(txin_type, privkey, compressed):

pubkey = ecc.ECPrivkey(privkey).get_public_key_hex(compressed=compressed)

_append_utxos_to_inputs(inputs, network, pubkey, txin_type, imax)

keypairs[pubkey] = privkey, compressed

inputs = [] # type: List[PartialTxInput]

keypairs = {}

for sec in privkeys:

txin_type, privkey, compressed = bitcoin.deserialize_privkey(sec)

find_utxos_for_privkey(txin_type, privkey, compressed)

# do other lookups to increase support coverage

if is_minikey(sec):

# minikeys don't have a compressed byte

# we lookup both compressed and uncompressed pubkeys

find_utxos_for_privkey(txin_type, privkey, not compressed)

elif txin_type == 'p2pkh':

# WIF serialization does not distinguish p2pkh and p2pk

# we also search for pay-to-pubkey outputs

find_utxos_for_privkey('p2pk', privkey, compressed)

if not inputs:

raise Exception(_('No inputs found. (Note that inputs need to be confirmed)'))

# FIXME actually inputs need not be confirmed now, see https://github.com/kyuupichan/electrumx/issues/365

to_address: str, fee: int = None, imax=100,

locktime=None, tx_version=None) -> PartialTransaction:

inputs, keypairs = sweep_preparations(privkeys, network, imax)

total = sum(txin.value_sats() for txin in inputs)

if fee is None:

outputs = [PartialTxOutput(scriptpubkey=bfh(bitcoin.address_to_script(to_address)),

value=total)]

tx = PartialTransaction.from_io(inputs, outputs)

fee = config.estimate_fee(tx.estimated_size())

if total - fee < 0:

raise Exception(_('Not enough funds on address.') + '\nTotal: %d satoshis\nFee: %d'%(total, fee))

if total - fee < dust_threshold(network):

raise Exception(_('Not enough funds on address.') + '\nTotal: %d satoshis\nFee: %d\nDust Threshold: %d'%(total, fee, dust_threshold(network)))

outputs = [PartialTxOutput(scriptpubkey=bfh(bitcoin.address_to_script(to_address)),

value=total - fee)]

if locktime is None:

locktime = get_locktime_for_new_transaction(network)

tx = PartialTransaction.from_io(inputs, outputs, locktime=locktime, version=tx_version)

tx.set_rbf(True)

tx.sign(keypairs)

# if no network or not up to date, just set locktime to zero

if not network:

return 0

chain = network.blockchain()

header = chain.header_at_tip()

if not header:

return 0

STALE_DELAY = 8 * 60 * 60 # in seconds

if header['timestamp'] + STALE_DELAY < time.time():

return 0

# discourage "fee sniping"

locktime = chain.height()

# sometimes pick locktime a bit further back, to help privacy

# of setups that need more time (offline/multisig/coinjoin/...)

if random.randint(0, 9) == 0:

locktime = max(0, locktime - random.randint(0, 99))

txid: Optional[str]

status: str

label: str

can_broadcast: bool

can_bump: bool

can_save_as_local: bool

amount: Optional[int]

fee: Optional[int]

tx_mined_status: TxMinedInfo

"""

LOGGING_SHORTCUT = 'w'

max_change_outputs = 3

gap_limit_for_change = 6

txin_type: str

wallet_type: str

def __init__(self, storage: WalletStorage, *, config: SimpleConfig):

if not storage.is_ready_to_be_used_by_wallet():

raise Exception("storage not ready to be used by Abstract_Wallet")

self.config = config

assert self.config is not None, "config must not be None"

self.storage = storage

# load addresses needs to be called before constructor for sanity checks

self.storage.db.load_addresses(self.wallet_type)

self.keystore = None # type: Optional[KeyStore] # will be set by load_keystore

AddressSynchronizer.__init__(self, storage.db)

# saved fields

self.use_change = storage.get('use_change', True)

self.multiple_change = storage.get('multiple_change', False)

self.labels = storage.db.get_dict('labels')

self.frozen_addresses = set(storage.get('frozen_addresses', []))

self.frozen_coins = set(storage.get('frozen_coins', [])) # set of txid:vout strings

self.fiat_value = storage.db.get_dict('fiat_value')

self.receive_requests = storage.db.get_dict('payment_requests')

self.invoices = storage.db.get_dict('invoices')

self._prepare_onchain_invoice_paid_detection()

self.calc_unused_change_addresses()

# save wallet type the first time

if self.storage.get('wallet_type') is None:

self.storage.put('wallet_type', self.wallet_type)

self.contacts = Contacts(self.storage)

self._coin_price_cache = {}

# lightning

ln_xprv = self.storage.get('lightning_privkey2')

self.lnworker = LNWallet(self, ln_xprv) if ln_xprv else None

def has_lightning(self):

return bool(self.lnworker)

def init_lightning(self):

if self.storage.get('lightning_privkey2'):

return

if not is_using_fast_ecc():

raise Exception('libsecp256k1 library not available. '

'Verifying Lightning channels is too computationally expensive without libsecp256k1, aborting.')

# TODO derive this deterministically from wallet.keystore at keystore generation time

# probably along a hardened path ( lnd-equivalent would be m/1017'/coinType'/ )

seed = os.urandom(32)

node = BIP32Node.from_rootseed(seed, xtype='standard')

ln_xprv = node.to_xprv()

self.storage.put('lightning_privkey2', ln_xprv)

self.storage.write()

def remove_lightning(self):

if not self.storage.get('lightning_privkey2'):

return

if bool(self.lnworker.channels):

raise Exception('Error: This wallet has channels')

self.storage.put('lightning_privkey2', None)

self.storage.write()

def stop_threads(self):

super().stop_threads()

if any([ks.is_requesting_to_be_rewritten_to_wallet_file for ks in self.get_keystores()]):

self.save_keystore()

self.storage.write()

def set_up_to_date(self, b):

super().set_up_to_date(b)

if b: self.storage.write()

def clear_history(self):

super().clear_history()

self.storage.write()

def start_network(self, network):

AddressSynchronizer.start_network(self, network)

if self.lnworker:

network.maybe_init_lightning()

self.lnworker.start_network(network)

def load_and_cleanup(self):

self.load_keystore()

self.test_addresses_sanity()

super().load_and_cleanup()

@abstractmethod

def load_keystore(self) -> None:

pass

def diagnostic_name(self):

return self.basename()

def __str__(self):

return self.basename()

def get_master_public_key(self):

return None

def get_master_public_keys(self):

return []

def basename(self) -> str:

return os.path.basename(self.storage.path)

def test_addresses_sanity(self) -> None:

addrs = self.get_receiving_addresses()

if len(addrs) > 0:

addr = str(addrs[0])

if not bitcoin.is_address(addr):

neutered_addr = addr[:5] + '..' + addr[-2:]

raise WalletFileException(f'The addresses in this wallet are not bitcoin addresses.\n'

f'e.g. {neutered_addr} (length: {len(addr)})')

def calc_unused_change_addresses(self):

with self.lock:

if hasattr(self, '_unused_change_addresses'):

addrs = self._unused_change_addresses

else:

addrs = self.get_change_addresses()

self._unused_change_addresses = [addr for addr in addrs if not self.is_used(addr)]

return list(self._unused_change_addresses)

def is_deterministic(self) -> bool:

return self.keystore.is_deterministic()

def set_label(self, name, text = None):

changed = False

old_text = self.labels.get(name)

if text:

text = text.replace("\n", " ")

if old_text != text:

self.labels[name] = text

changed = True

else:

if old_text is not None:

self.labels.pop(name)

changed = True

if changed:

run_hook('set_label', self, name, text)

#self.storage.put('labels', self.labels)

self.storage.write()

return changed

def set_fiat_value(self, txid, ccy, text, fx, value_sat):

if not self.db.get_transaction(txid):

return

# since fx is inserting the thousands separator,

# and not util, also have fx remove it

text = fx.remove_thousands_separator(text)

def_fiat = self.default_fiat_value(txid, fx, value_sat)

formatted = fx.ccy_amount_str(def_fiat, commas=False)

def_fiat_rounded = Decimal(formatted)

reset = not text

if not reset:

try:

text_dec = Decimal(text)

text_dec_rounded = Decimal(fx.ccy_amount_str(text_dec, commas=False))

reset = text_dec_rounded == def_fiat_rounded

except:

# garbage. not resetting, but not saving either

return False

if reset:

d = self.fiat_value.get(ccy, {})

if d and txid in d:

d.pop(txid)

else:

# avoid saving empty dict

return True

else:

if ccy not in self.fiat_value:

self.fiat_value[ccy] = {}

self.fiat_value[ccy][txid] = text

return reset

def get_fiat_value(self, txid, ccy):

fiat_value = self.fiat_value.get(ccy, {}).get(txid)

try:

return Decimal(fiat_value)

except:

return

def is_mine(self, address) -> bool:

return bool(self.get_address_index(address))

def is_change(self, address) -> bool:

if not self.is_mine(address):

return False

return self.get_address_index(address)[0] == 1

@abstractmethod

def get_address_index(self, address: str) -> Optional[AddressIndexGeneric]:

pass

@abstractmethod

def get_redeem_script(self, address: str) -> Optional[str]:

pass

@abstractmethod

def get_witness_script(self, address: str) -> Optional[str]:

pass

@abstractmethod

def get_txin_type(self, address: str) -> str:

"""Return script type of wallet address."""

pass

def export_private_key(self, address, password) -> str:

if self.is_watching_only():

raise Exception(_("This is a watching-only wallet"))

if not is_address(address):

raise Exception(f"Invalid bitcoin address: {address}")

if not self.is_mine(address):

raise Exception(_('Address not in wallet.') + f' {address}')

index = self.get_address_index(address)

pk, compressed = self.keystore.get_private_key(index, password)

txin_type = self.get_txin_type(address)

serialized_privkey = bitcoin.serialize_privkey(pk, compressed, txin_type)

return serialized_privkey

@abstractmethod

def get_public_keys(self, address: str) -> Sequence[str]:

pass

def get_public_keys_with_deriv_info(self, address: str) -> Dict[bytes, Tuple[KeyStoreWithMPK, Sequence[int]]]:

"""Returns a map: pubkey -> (keystore, derivation_suffix)"""

return {}

def get_tx_info(self, tx) -> TxWalletDetails:

is_relevant, is_mine, v, fee = self.get_wallet_delta(tx)

if fee is None and isinstance(tx, PartialTransaction):

fee = tx.get_fee()

exp_n = None

can_broadcast = False

can_bump = False

can_save_as_local = False

label = ''

tx_hash = tx.txid()

tx_mined_status = self.get_tx_height(tx_hash)

if tx.is_complete():

if self.db.get_transaction(tx_hash):

label = self.get_label(tx_hash)

if tx_mined_status.height > 0:

if tx_mined_status.conf:

status = _("{} confirmations").format(tx_mined_status.conf)

else:

status = _('Not verified')

elif tx_mined_status.height in (TX_HEIGHT_UNCONF_PARENT, TX_HEIGHT_UNCONFIRMED):

status = _('Unconfirmed')

if fee is None:

fee = self.get_tx_fee(tx_hash)

if fee and self.network and self.config.has_fee_mempool():

size = tx.estimated_size()

fee_per_byte = fee / size

exp_n = self.config.fee_to_depth(fee_per_byte)

can_bump = is_mine and not tx.is_final()

else:

status = _('Local')

can_broadcast = self.network is not None

can_bump = is_mine and not tx.is_final()

else:

status = _("Signed")

can_broadcast = self.network is not None

can_save_as_local = is_relevant

else:

s, r = tx.signature_count()

status = _("Unsigned") if s == 0 else _('Partially signed') + ' (%d/%d)'%(s,r)

if is_relevant:

if is_mine:

if fee is not None:

amount = v + fee

else:

amount = v

else:

amount = v

else:

amount = None

return TxWalletDetails(

txid=tx_hash,

status=status,

label=label,

can_broadcast=can_broadcast,

can_bump=can_bump,

can_save_as_local=can_save_as_local,

amount=amount,

fee=fee,

tx_mined_status=tx_mined_status,

mempool_depth_bytes=exp_n,

)

def get_spendable_coins(self, domain, *, nonlocal_only=False) -> Sequence[PartialTxInput]:

confirmed_only = self.config.get('confirmed_only', False)

utxos = self.get_utxos(domain,

excluded_addresses=self.frozen_addresses,

mature_only=True,

confirmed_only=confirmed_only,

nonlocal_only=nonlocal_only)

utxos = [utxo for utxo in utxos if not self.is_frozen_coin(utxo)]

return utxos

@abstractmethod

def get_receiving_addresses(self, *, slice_start=None, slice_stop=None) -> Sequence[str]:

pass

@abstractmethod

def get_change_addresses(self, *, slice_start=None, slice_stop=None) -> Sequence[str]:

pass

def dummy_address(self):

# first receiving address

return self.get_receiving_addresses(slice_start=0, slice_stop=1)[0]

def get_frozen_balance(self):

if not self.frozen_coins: # shortcut

return self.get_balance(self.frozen_addresses)

c1, u1, x1 = self.get_balance()

c2, u2, x2 = self.get_balance(excluded_addresses=self.frozen_addresses,

excluded_coins=self.frozen_coins)

return c1-c2, u1-u2, x1-x2

def balance_at_timestamp(self, domain, target_timestamp):

# we assume that get_history returns items ordered by block height

# we also assume that block timestamps are monotonic (which is false...!)

h = self.get_history(domain=domain)

balance = 0

for hist_item in h:

balance = hist_item.balance

if hist_item.tx_mined_status.timestamp is None or hist_item.tx_mined_status.timestamp > target_timestamp:

return balance - hist_item.delta

# return last balance

return balance

def get_onchain_history(self, *, domain=None):

for hist_item in self.get_history(domain=domain):

yield {

'txid': hist_item.txid,

'fee_sat': hist_item.fee,

'height': hist_item.tx_mined_status.height,

'confirmations': hist_item.tx_mined_status.conf,

'timestamp': hist_item.tx_mined_status.timestamp,

'incoming': True if hist_item.delta>0 else False,

'bc_value': Satoshis(hist_item.delta),

'bc_balance': Satoshis(hist_item.balance),

'date': timestamp_to_datetime(hist_item.tx_mined_status.timestamp),

'label': self.get_label(hist_item.txid),

'txpos_in_block': hist_item.tx_mined_status.txpos,

}

def create_invoice(self, outputs: List[PartialTxOutput], message, pr, URI):

if '!' in (x.value for x in outputs):

amount = '!'

else:

amount = sum(x.value for x in outputs)

invoice = {

'type': PR_TYPE_ONCHAIN,

'message': message,

'outputs': outputs,

'amount': amount,

}

if pr:

invoice['bip70'] = pr.raw.hex()

invoice['time'] = pr.get_time()

invoice['exp'] = pr.get_expiration_date() - pr.get_time()

invoice['requestor'] = pr.get_requestor()

invoice['message'] = pr.get_memo()

elif URI:

timestamp = URI.get('time')

if timestamp: invoice['time'] = timestamp

exp = URI.get('exp')

if exp: invoice['exp'] = exp

if 'time' not in invoice:

invoice['time'] = int(time.time())

return invoice

def save_invoice(self, invoice):

invoice_type = invoice['type']

if invoice_type == PR_TYPE_LN:

key = invoice['rhash']

elif invoice_type == PR_TYPE_ONCHAIN:

key = bh2u(sha256(repr(invoice))[0:16])

invoice['id'] = key

outputs = invoice['outputs'] # type: List[PartialTxOutput]

with self.transaction_lock:

for txout in outputs:

self._invoices_from_scriptpubkey_map[txout.scriptpubkey].add(key)

else:

raise Exception('Unsupported invoice type')

self.invoices[key] = invoice

self.storage.write()

def clear_invoices(self):

self.invoices = {}

self.storage.write()

def get_invoices(self):

out = [self.get_invoice(key) for key in self.invoices.keys()]

out = list(filter(None, out))

out.sort(key=operator.itemgetter('time'))

return out

def get_invoice(self, key):

if key not in self.invoices:

return

# convert StorageDict to dict

item = dict(self.invoices[key])

request_type = item.get('type')

if request_type == PR_TYPE_ONCHAIN:

item['status'] = PR_PAID if self._is_onchain_invoice_paid(item)[0] else PR_UNPAID

elif self.lnworker and request_type == PR_TYPE_LN:

item['status'] = self.lnworker.get_payment_status(bfh(item['rhash']))

else:

return

return item

def _get_relevant_invoice_keys_for_tx(self, tx: Transaction) -> Set[str]:

relevant_invoice_keys = set()

for txout in tx.outputs():

for invoice_key in self._invoices_from_scriptpubkey_map.get(txout.scriptpubkey, set()):

relevant_invoice_keys.add(invoice_key)

return relevant_invoice_keys

def _prepare_onchain_invoice_paid_detection(self):

# scriptpubkey -> list(invoice_keys)

self._invoices_from_scriptpubkey_map = defaultdict(set) # type: Dict[bytes, Set[str]]

for invoice_key, invoice in self.invoices.items():

if invoice.get('type') == PR_TYPE_ONCHAIN:

outputs = invoice['outputs'] # type: List[PartialTxOutput]

for txout in outputs:

self._invoices_from_scriptpubkey_map[txout.scriptpubkey].add(invoice_key)

def _is_onchain_invoice_paid(self, invoice) -> Tuple[bool, Sequence[str]]:

"""Returns whether on-chain invoice is satisfied, and list of relevant TXIDs."""

assert invoice.get('type') == PR_TYPE_ONCHAIN

invoice_amounts = defaultdict(int) # type: Dict[bytes, int] # scriptpubkey -> value_sats

for txo in invoice['outputs']: # type: PartialTxOutput

invoice_amounts[txo.scriptpubkey] += 1 if txo.value == '!' else txo.value

relevant_txs = []

with self.transaction_lock:

for invoice_scriptpubkey, invoice_amt in invoice_amounts.items():

scripthash = bitcoin.script_to_scripthash(invoice_scriptpubkey.hex())

prevouts_and_values = self.db.get_prevouts_by_scripthash(scripthash)

relevant_txs += [prevout.txid.hex() for prevout, v in prevouts_and_values]

total_received = sum([v for prevout, v in prevouts_and_values])

if total_received < invoice_amt:

return False, []

return True, relevant_txs

def _maybe_set_tx_label_based_on_invoices(self, tx: Transaction) -> bool:

tx_hash = tx.txid()

with self.transaction_lock:

labels = []

for invoice_key in self._get_relevant_invoice_keys_for_tx(tx):

invoice = self.invoices.get(invoice_key)

if invoice is None: continue

assert invoice.get('type') == PR_TYPE_ONCHAIN

if invoice['message']:

labels.append(invoice['message'])

if labels:

self.set_label(tx_hash, "; ".join(labels))

return bool(labels)

def add_transaction(self, tx, *, allow_unrelated=False):

tx_was_added = super().add_transaction(tx, allow_unrelated=allow_unrelated)

if tx_was_added:

self._maybe_set_tx_label_based_on_invoices(tx)

return tx_was_added

@profiler

def get_full_history(self, fx=None, *, onchain_domain=None, include_lightning=True):

transactions = OrderedDictWithIndex()

onchain_history = self.get_onchain_history(domain=onchain_domain)

for tx_item in onchain_history:

txid = tx_item['txid']

transactions[txid] = tx_item

if self.lnworker and include_lightning:

lightning_history = self.lnworker.get_history()

else:

lightning_history = []

for i, tx_item in enumerate(lightning_history):

txid = tx_item.get('txid')

ln_value = Decimal(tx_item['amount_msat']) / 1000

if txid and txid in transactions:

item = transactions[txid]

item['label'] = tx_item['label']

item['ln_value'] = Satoshis(ln_value)

item['ln_balance_msat'] = tx_item['balance_msat']

else:

tx_item['lightning'] = True

tx_item['ln_value'] = Satoshis(ln_value)

tx_item['txpos'] = i # for sorting

key = tx_item['payment_hash'] if 'payment_hash' in tx_item else tx_item['type'] + tx_item['channel_id']

transactions[key] = tx_item

now = time.time()

balance = 0

for item in transactions.values():

# add on-chain and lightning values

value = Decimal(0)

if item.get('bc_value'):

value += item['bc_value'].value

if item.get('ln_value'):

value += item.get('ln_value').value

item['value'] = Satoshis(value)

balance += value

item['balance'] = Satoshis(balance)

if fx:

timestamp = item['timestamp'] or now

fiat_value = value / Decimal(bitcoin.COIN) * fx.timestamp_rate(timestamp)

item['fiat_value'] = Fiat(fiat_value, fx.ccy)

item['fiat_default'] = True

return transactions

@profiler

def get_detailed_history(self, from_timestamp=None, to_timestamp=None,

fx=None, show_addresses=False):

# History with capital gains, using utxo pricing

# FIXME: Lightning capital gains would requires FIFO

out = []

income = 0

expenditures = 0

capital_gains = Decimal(0)

fiat_income = Decimal(0)

fiat_expenditures = Decimal(0)

now = time.time()

for item in self.get_onchain_history():

timestamp = item['timestamp']

if from_timestamp and (timestamp or now) < from_timestamp:

continue

if to_timestamp and (timestamp or now) >= to_timestamp:

continue

tx_hash = item['txid']

tx = self.db.get_transaction(tx_hash)

tx_fee = item['fee_sat']

item['fee'] = Satoshis(tx_fee) if tx_fee is not None else None

if show_addresses:

item['inputs'] = list(map(lambda x: x.to_json(), tx.inputs()))

item['outputs'] = list(map(lambda x: {'address': x.get_ui_address_str(), 'value': Satoshis(x.value)},

tx.outputs()))

# fixme: use in and out values

value = item['bc_value'].value

if value < 0:

expenditures += -value

else:

income += value

# fiat computations

if fx and fx.is_enabled() and fx.get_history_config():

fiat_fields = self.get_tx_item_fiat(tx_hash, value, fx, tx_fee)

fiat_value = fiat_fields['fiat_value'].value

item.update(fiat_fields)

if value < 0:

capital_gains += fiat_fields['capital_gain'].value

fiat_expenditures += -fiat_value

else:

fiat_income += fiat_value

out.append(item)

# add summary

if out:

b, v = out[0]['bc_balance'].value, out[0]['bc_value'].value

start_balance = None if b is None or v is None else b - v

end_balance = out[-1]['bc_balance'].value

if from_timestamp is not None and to_timestamp is not None:

start_date = timestamp_to_datetime(from_timestamp)

end_date = timestamp_to_datetime(to_timestamp)

else:

start_date = None

end_date = None

summary = {

'start_date': start_date,

'end_date': end_date,

'start_balance': Satoshis(start_balance),

'end_balance': Satoshis(end_balance),

'incoming': Satoshis(income),

'outgoing': Satoshis(expenditures)

}

if fx and fx.is_enabled() and fx.get_history_config():

unrealized = self.unrealized_gains(None, fx.timestamp_rate, fx.ccy)

summary['fiat_currency'] = fx.ccy

summary['fiat_capital_gains'] = Fiat(capital_gains, fx.ccy)

summary['fiat_incoming'] = Fiat(fiat_income, fx.ccy)

summary['fiat_outgoing'] = Fiat(fiat_expenditures, fx.ccy)

summary['fiat_unrealized_gains'] = Fiat(unrealized, fx.ccy)

summary['fiat_start_balance'] = Fiat(fx.historical_value(start_balance, start_date), fx.ccy)

summary['fiat_end_balance'] = Fiat(fx.historical_value(end_balance, end_date), fx.ccy)

summary['fiat_start_value'] = Fiat(fx.historical_value(COIN, start_date), fx.ccy)

summary['fiat_end_value'] = Fiat(fx.historical_value(COIN, end_date), fx.ccy)

else:

summary = {}

return {

'transactions': out,

'summary': summary

}

def default_fiat_value(self, tx_hash, fx, value_sat):

return value_sat / Decimal(COIN) * self.price_at_timestamp(tx_hash, fx.timestamp_rate)

def get_tx_item_fiat(self, tx_hash, value, fx, tx_fee):

item = {}

fiat_value = self.get_fiat_value(tx_hash, fx.ccy)

fiat_default = fiat_value is None

fiat_rate = self.price_at_timestamp(tx_hash, fx.timestamp_rate)

fiat_value = fiat_value if fiat_value is not None else self.default_fiat_value(tx_hash, fx, value)

fiat_fee = tx_fee / Decimal(COIN) * fiat_rate if tx_fee is not None else None

item['fiat_currency'] = fx.ccy

item['fiat_rate'] = Fiat(fiat_rate, fx.ccy)

item['fiat_value'] = Fiat(fiat_value, fx.ccy)

item['fiat_fee'] = Fiat(fiat_fee, fx.ccy) if fiat_fee else None

item['fiat_default'] = fiat_default

if value < 0:

acquisition_price = - value / Decimal(COIN) * self.average_price(tx_hash, fx.timestamp_rate, fx.ccy)

liquidation_price = - fiat_value

item['acquisition_price'] = Fiat(acquisition_price, fx.ccy)

cg = liquidation_price - acquisition_price

item['capital_gain'] = Fiat(cg, fx.ccy)

return item

def get_label(self, tx_hash: str) -> str:

return self.labels.get(tx_hash, '') or self.get_default_label(tx_hash)

def get_default_label(self, tx_hash) -> str:

if not self.db.get_txi_addresses(tx_hash):

labels = []

for addr in self.db.get_txo_addresses(tx_hash):

label = self.labels.get(addr)

if label:

labels.append(label)

return ', '.join(labels)

return ''

def get_tx_status(self, tx_hash, tx_mined_info: TxMinedInfo):

extra = []

height = tx_mined_info.height

conf = tx_mined_info.conf

timestamp = tx_mined_info.timestamp

if height == TX_HEIGHT_FUTURE:

assert conf < 0, conf

num_blocks_remainining = -conf

return 2, f'in {num_blocks_remainining} blocks'

if conf == 0:

tx = self.db.get_transaction(tx_hash)

if not tx:

return 2, 'unknown'

is_final = tx and tx.is_final()

if not is_final:

extra.append('rbf')

fee = self.get_tx_fee(tx_hash)

if fee is not None:

size = tx.estimated_size()

fee_per_byte = fee / size

extra.append(format_fee_satoshis(fee_per_byte) + ' sat/b')

if fee is not None and height in (TX_HEIGHT_UNCONF_PARENT, TX_HEIGHT_UNCONFIRMED) \

and self.config.has_fee_mempool():

exp_n = self.config.fee_to_depth(fee_per_byte)

if exp_n:

extra.append('%.2f MB'%(exp_n/1000000))

if height == TX_HEIGHT_LOCAL:

status = 3

elif height == TX_HEIGHT_UNCONF_PARENT:

status = 1

elif height == TX_HEIGHT_UNCONFIRMED:

status = 0

else:

status = 2 # not SPV verified

else:

status = 3 + min(conf, 6)

time_str = format_time(timestamp) if timestamp else _("unknown")

status_str = TX_STATUS[status] if status < 4 else time_str

if extra:

status_str += ' [%s]'%(', '.join(extra))

return status, status_str

def relayfee(self):

return relayfee(self.network)

def dust_threshold(self):

return dust_threshold(self.network)

def get_unconfirmed_base_tx_for_batching(self) -> Optional[Transaction]:

candidate = None

for hist_item in self.get_history():

# tx should not be mined yet

if hist_item.tx_mined_status.conf > 0: continue

# conservative future proofing of code: only allow known unconfirmed types

if hist_item.tx_mined_status.height not in (TX_HEIGHT_UNCONFIRMED,

TX_HEIGHT_UNCONF_PARENT,

TX_HEIGHT_LOCAL):

continue

# tx should be "outgoing" from wallet

if hist_item.delta >= 0:

continue

tx = self.db.get_transaction(hist_item.txid)

if not tx:

continue

# is_mine outputs should not be spent yet

# to avoid cancelling our own dependent transactions

txid = tx.txid()

if any([self.is_mine(o.address) and self.db.get_spent_outpoint(txid, output_idx)

for output_idx, o in enumerate(tx.outputs())]):

continue

# all inputs should be is_mine

if not all([self.is_mine(self.get_txin_address(txin)) for txin in tx.inputs()]):

continue

# prefer txns already in mempool (vs local)

if hist_item.tx_mined_status.height == TX_HEIGHT_LOCAL:

candidate = tx

continue

# tx must have opted-in for RBF

if tx.is_final(): continue

return tx

return candidate

def get_change_addresses_for_new_transaction(self, preferred_change_addr=None) -> List[str]:

change_addrs = []

if preferred_change_addr:

if isinstance(preferred_change_addr, (list, tuple)):

change_addrs = list(preferred_change_addr)

else:

change_addrs = [preferred_change_addr]

elif self.use_change:

# Recalc and get unused change addresses

addrs = self.calc_unused_change_addresses()

# New change addresses are created only after a few

# confirmations.

if addrs:

# if there are any unused, select all

change_addrs = addrs

else:

# if there are none, take one randomly from the last few

addrs = self.get_change_addresses(slice_start=-self.gap_limit_for_change)

change_addrs = [random.choice(addrs)] if addrs else []

for addr in change_addrs:

assert is_address(addr), f"not valid bitcoin address: {addr}"