diff --git a/controller.py b/controller.py index f09ecf6..ae71c0e 100644 --- a/controller.py +++ b/controller.py @@ -1,12 +1,14 @@ import multiprocessing, sys import jobs, rebalancer, htlc_stream, p2p, manage +import logging +logger = logging.getLogger('[Controller]') def run_task(task): task() def main(): tasks = [jobs.main, rebalancer.main, htlc_stream.main, p2p.main] - print('Controller is starting...') + logger.info('Starting all LNDg processes...') processes = [] for task in tasks: @@ -15,15 +17,14 @@ def main(): process.start() if len(sys.argv) > 1: - sys.argv[0] = "manage.py" - process = multiprocessing.Process(target=manage.main(sys.argv), name="manage.py") + sys.argv[0] = 'manage.py' + process = multiprocessing.Process(target=manage.main(sys.argv), name='manage.py') processes.append(process) process.start() for process in processes: process.join() - - print('Controller is stopping...') + logger.info('Stopping all LNDg processes...') if __name__ == '__main__': main() diff --git a/gui/views.py b/gui/views.py index 1cb7512..95cb64c 100644 --- a/gui/views.py +++ b/gui/views.py @@ -29,6 +29,8 @@ from requests import get from secrets import token_bytes from trade import create_trade_details import af +import logging +logger = logging.getLogger('django.lndg') def graph_links(): if LocalSettings.objects.filter(key='GUI-GraphLinks').exists(): @@ -409,12 +411,10 @@ def closures(request): return redirect('home') def find_next_block_maturity(force_closing_channel): - #print (f"{datetime.now().strftime('%c')} : {force_closing_channel=}") if force_closing_channel.blocks_til_maturity > 0: return force_closing_channel.blocks_til_maturity for pending_htlc in force_closing_channel.pending_htlcs: if pending_htlc.blocks_til_maturity > 0: - #print (f"{datetime.now().strftime('%c')} : {pending_htlc=}") return pending_htlc.blocks_til_maturity return -1 @@ -1113,7 +1113,7 @@ def unprofitable_channels(request): stub = lnrpc.LightningStub(lnd_connect()) current_block_height = stub.GetInfo(ln.GetInfoRequest()).block_height except Exception as e: - print(f"Error getting current block height: {e}") + logger.error(f'Error getting current block height: {str(e)}') current_block_height = 0 VERY_NEW_DAYS = 7 @@ -1350,7 +1350,7 @@ def unprofitable_channels(request): except Exception as e: # Handle potential errors during age calculation gracefully - print(f"Error calculating age for chan_id {channel.chan_id}: {e}") + logger.error(f'Error calculating age for chan_id {channel.chan_id}: {str(e)}') channel_age_days = -1 # Indicate unknown age else: channel_age_days = -1 # Indicate unknown age if block height or chan_id missing @@ -1429,26 +1429,25 @@ def actions(request): result['auto_rebalance'] = channel.auto_rebalance result['ar_target'] = channel.ar_in_target if result['o7D'] > (result['i7D']*1.10) and result['outbound_percent'] > 75: - #print('Case 1: Pass') - continue + logger.debug('Auto-Enable Case 1: Pass') elif result['o7D'] > (result['i7D']*1.10) and result['inbound_percent'] > 75 and channel.auto_rebalance == False: if channel.local_fee_rate <= channel.remote_fee_rate: - #print('Case 6: Peer Fee Too High') + logger.debug('Case 6: Peer Fee Too High') result['output'] = 'Peer Fee Too High' result['reason'] = 'o7D > i7D AND Inbound Liq > 75% AND Local Fee < Remote Fee' continue - #print('Case 2: Enable AR') + logger.debug('Case 2: Enable AR - o7D > i7D AND Inbound Liq > 75%') result['output'] = 'Enable AR' result['reason'] = 'o7D > i7D AND Inbound Liq > 75%' elif result['o7D'] < (result['i7D']*1.10) and result['outbound_percent'] > 75 and channel.auto_rebalance == True: - #print('Case 3: Disable AR') + logger.debug('Case 3: Disable AR - o7D < i7D AND Outbound Liq > 75%') result['output'] = 'Disable AR' result['reason'] = 'o7D < i7D AND Outbound Liq > 75%' elif result['o7D'] < (result['i7D']*1.10) and result['inbound_percent'] > 75: - #print('Case 4: Pass') + logger.debug('Case 4: Pass') continue else: - #print('Case 5: Pass') + logger.debug('Case 5: Pass') continue if len(result) > 0: action_list.append(result) @@ -1725,7 +1724,7 @@ def batch_open(request): channel_open.local_funding_amount = open['amt'] channels.append(channel_open) response = stub.BatchOpenChannel(ln.BatchOpenChannelRequest(channels=channels, sat_per_vbyte=form.cleaned_data['fee_rate'])) - print(response) + logger.debug(f'Batch open response: {response}') messages.success(request, 'Batch opened channels!') except Exception as e: error = str(e) @@ -2777,7 +2776,7 @@ def get_channeldb_file_size(): # Check for required settings if not host_value or not user_value: - print("Error: Remote file size enabled, but host or user is not set.") + logger.error('Error: Remote file size enabled, but host or user is not set') return round(path.getsize(path.expanduser(settings.LND_DATABASE_PATH))*0.000000001, 3) # --- Paramiko logic --- @@ -2805,7 +2804,7 @@ def get_channeldb_file_size(): return round(file_size_bytes * 0.000000001, 3) except Exception as e: - print(f"Error retrieving file size with paramiko: {e}") + logger.error(f'Error retrieving file size with paramiko: {str(e)}') return round(path.getsize(path.expanduser(settings.LND_DATABASE_PATH))*0.000000001, 3) # Fallback # --- End Paramiko logic --- @@ -2815,7 +2814,7 @@ def get_channeldb_file_size(): except Exception as e: # Handle exceptions - print(f"Error retrieving channel.db file size: {e}") + logger.error(f'Error retrieving channel.db file size: {str(e)}') return 0 @api_view(['GET']) diff --git a/htlc_stream.py b/htlc_stream.py index e10d4ae..ad9185e 100644 --- a/htlc_stream.py +++ b/htlc_stream.py @@ -1,5 +1,4 @@ import django -from datetime import datetime from gui.lnd_deps import router_pb2 as lnr from gui.lnd_deps import router_pb2_grpc as lnrouter from gui.lnd_deps.lnd_connect import lnd_connect @@ -8,11 +7,13 @@ from time import sleep environ['DJANGO_SETTINGS_MODULE'] = 'lndg.settings' django.setup() from gui.models import Channels, FailedHTLCs +import logging +logger = logging.getLogger('[HTLC]') def main(): while True: try: - print(f"{datetime.now().strftime('%c')} : [HTLC] : Starting failed HTLC stream...") + logger.info('Starting failed HTLC stream...') connection = lnd_connect() routerstub = lnrouter.RouterStub(connection) all_forwards = {} @@ -59,7 +60,7 @@ def main(): FailedHTLCs(amount=amount, chan_id_in=in_chan_id, chan_id_out=out_chan_id, chan_in_alias=in_chan_alias, chan_out_alias=out_chan_alias, chan_out_liq=out_chan_liq, chan_out_pending=out_chan_pending, wire_failure=wire_failure, failure_detail=failure_detail, missed_fee=missed_fee).save() del all_forwards[key] except Exception as e: - print(f"{datetime.now().strftime('%c')} : [HTLC] : Error while running failed HTLC stream: {str(e)}") + logger.error(f'Error while running failed HTLC stream: {str(e)}') sleep(20) if __name__ == '__main__': diff --git a/initialize.py b/initialize.py index b2ff07e..f884d07 100644 --- a/initialize.py +++ b/initialize.py @@ -171,6 +171,76 @@ STATIC_URL = 'static/' STATIC_ROOT = os.path.join(BASE_DIR, 'gui/static/') DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField' SESSION_COOKIE_AGE = %s + +LOGGING = { + 'version': 1, + 'disable_existing_loggers': False, + 'formatters': { + 'verbose': { + 'format': '{asctime} {name} {levelname} - {message}', + 'style': '{', + }, + 'simple': { + 'format': '{asctime} {name} - {message}', + 'style': '{', + }, + }, + 'handlers': { + 'app-file': { + 'level': 'INFO', + 'class': 'logging.handlers.RotatingFileHandler', + 'filename': 'data/lndg-controller.log', + 'maxBytes': 25*(1024*1024), + 'backupCount': 5, + 'formatter': 'verbose', + }, + 'web-file': { + 'level': 'INFO', + 'class': 'logging.handlers.RotatingFileHandler', + 'filename': 'data/lndg-web.log', + 'maxBytes': 25*(1024*1024), + 'backupCount': 5, + 'formatter': 'verbose', + }, + 'console': { + 'level': 'INFO', + 'class': 'logging.StreamHandler', + 'formatter': 'simple', + }, + }, + 'loggers': { + '[Controller]': { + 'handlers': ['app-file', 'console'], + 'level': 'INFO', + 'propagate': False, + }, + '[Data]': { + 'handlers': ['app-file', 'console'], + 'level': 'INFO', + 'propagate': False, + }, + '[Rebalancer]': { + 'handlers': ['app-file', 'console'], + 'level': 'INFO', + 'propagate': False, + }, + '[HTLC]': { + 'handlers': ['app-file', 'console'], + 'level': 'INFO', + 'propagate': False, + }, + '[P2P]': { + 'handlers': ['app-file', 'console'], + 'level': 'INFO', + 'propagate': False, + }, + 'django': { + 'handlers': ['web-file', 'console'], + 'level': 'INFO', + 'propagate': False, + }, + }, +} ''' % (secret, debug, node_ip, csrf, lnd_tls_path, lnd_macaroon_path, lnd_database_path, lnd_network, lnd_rpc_server, lnd_max_message, not nologinrequired, wnl, api_login, cookie_age) if not force_new and Path("lndg/settings.py").exists(): print('A settings file already exist, skipping creation...') diff --git a/jobs.py b/jobs.py index a779628..e3b5f00 100644 --- a/jobs.py +++ b/jobs.py @@ -15,7 +15,8 @@ environ['DJANGO_SETTINGS_MODULE'] = 'lndg.settings' django.setup() from gui.models import Payments, PaymentHops, Invoices, Forwards, Channels, Peers, Onchain, Closures, Resolutions, PendingHTLCs, LocalSettings, FailedHTLCs, Autofees, InboundFeeLog, PendingChannels, HistFailedHTLC, PeerEvents import af - +import logging +logger = logging.getLogger('[Data]') _SELF_PUBKEY = None def _get_self_pubkey(stub): @@ -24,7 +25,7 @@ def _get_self_pubkey(stub): try: _SELF_PUBKEY = stub.GetInfo(ln.GetInfoRequest()).identity_pubkey except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Data] : GetInfo failed: {e}") + logger.critical(f'GetInfo failed to get self pubkey: {e}') raise return _SELF_PUBKEY @@ -33,7 +34,7 @@ def update_payments(stub): inflight_payments = Payments.objects.filter(status=1).order_by('index') for payment in inflight_payments: payment_data = stub.ListPayments(ln.ListPaymentsRequest(include_incomplete=True, index_offset=payment.index-1, max_payments=1)).payments - #Ignore inflight payments before 30 days + # Ignore inflight payments before 30 days if len(payment_data) > 0 and payment.payment_hash == payment_data[0].payment_hash and payment.creation_date > (datetime.now() - timedelta(days=30)): update_payment(stub, payment_data[0], self_pubkey) else: @@ -79,7 +80,7 @@ def update_payments(stub): try: Payments.objects.bulk_create(new_payments, ignore_conflicts=True, batch_size=1000) except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Data] : Error bulk inserting payments: {str(e)}") + logger.error(f'Error bulk inserting payments: {str(e)}') for p in payments: if (p.payment_hash in new_hashes) or (p.status in (0, 1)): @@ -127,7 +128,7 @@ def update_payments(stub): try: Payments.objects.bulk_create(new_payments, ignore_conflicts=True, batch_size=1000) except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Data] : Error bulk inserting payments: {str(e)}") + logger.error(f'Error bulk inserting payments: {str(e)}') for p in payments: if (p.payment_hash in new_hashes) or (p.status in (0, 1)): @@ -214,7 +215,7 @@ def update_invoice(stub, invoice, db_invoice): try: valid = signerstub.VerifyMessage(lns.VerifyMessageReq(msg=(records[34349339]+bytes.fromhex(self_pubkey)+records[34349343]+records[34349334]), signature=records[34349337], pubkey=records[34349339])).valid except: - print(f"{datetime.now().strftime('%c')} : [Data] : Unable to validate signature on invoice: {invoice.r_hash.hex()}") + logger.error(f'Unable to validate signature on invoice: {invoice.r_hash.hex()}') valid = False sender = records[34349339].hex() if valid == True else None try: @@ -286,11 +287,11 @@ def update_forwards(stub): def disconnectpeer(stub, peer): try: stub.DisconnectPeer(ln.DisconnectPeerRequest(pub_key=peer.pubkey)) - print(f"{datetime.now().strftime('%c')} : [Data] : Disconnected peer {peer.alias} {peer.pubkey}") + logger.info(f'Disconnected peer {peer.alias} {peer.pubkey}') peer.connected = False peer.save() except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Data] : Error disconnecting peer {peer.alias} {peer.pubkey}: {str(e)}") + logger.error(f'Error disconnecting peer {peer.alias} {peer.pubkey}: {str(e)}') def update_channels(stub): counter = 0 @@ -302,11 +303,11 @@ def update_channels(stub): version = get_info.version for channel in channels: if Channels.objects.filter(chan_id=channel.chan_id).exists(): - #Update the channel record with the most current data + # Update the channel record with the most current data db_channel = Channels.objects.filter(chan_id=channel.chan_id)[0] pending_channel = None else: - #Create a record for this new channel + # Create a record for this new channel try: alias = stub.GetNodeInfo(ln.NodeInfoRequest(pub_key=channel.remote_pubkey, include_channels=False)).node.alias except: @@ -359,12 +360,12 @@ def update_channels(stub): if htlc.expiration_height - block_height <= 13: # If htlc is expiring within 13 blocks, disconnect peer to help resolve the stuck htlc peer = Peers.objects.filter(pubkey=channel.remote_pubkey)[0] if Peers.objects.filter(pubkey=channel.remote_pubkey).exists() else None if peer and (not peer.last_reconnected or (int((datetime.now() - peer.last_reconnected).total_seconds() / 60) > 10)): - print(f"{datetime.now().strftime('%c')} : [Data] : HTLC expiring at {htlc.expiration_height} and within 13 blocks of {block_height}, disconnecting peer {channel.remote_pubkey} to resolve HTLC: {htlc.hash_lock.hex()} ") + logger.info(f'HTLC expiring at {htlc.expiration_height} and within 13 blocks of {block_height}, disconnecting peer {channel.remote_pubkey} to resolve HTLC: {htlc.hash_lock.hex()}') disconnectpeer(stub, peer) peer.last_reconnected = datetime.now() peer.save() else: - print(f"{datetime.now().strftime('%c')} : [Data] : Could not find peer {channel.remote_pubkey} with expiring HTLC: {htlc.hash_lock.hex()}") + logger.error(f'Could not find peer {channel.remote_pubkey} with expiring HTLC: {htlc.hash_lock.hex()}') db_channel.pending_outbound = pending_out db_channel.pending_inbound = pending_in db_channel.htlc_count = htlc_counter @@ -474,9 +475,9 @@ def update_channels(stub): db_channel.remote_inbound_base_fee = 0 db_channel.remote_inbound_fee_rate = 0 except Exception as e: # LND has not found the channel on the graph - print(f"{datetime.now().strftime('%c')} : [Data] : Error getting graph data for channel {db_channel.chan_id}: {str(e)}") + logger.error(f'Error getting graph data for channel {db_channel.chan_id}: {str(e)}') if pending_channel: # skip adding new channel to the list, LND may not have added to the graph yet - print(f"{datetime.now().strftime('%c')} : [Data] : Waiting for pending channel {db_channel.chan_id} to be added to the graph...") + logger.error(f'Waiting for pending channel {db_channel.chan_id} to be added to the graph...') continue else: old_fee_rate = None @@ -525,16 +526,16 @@ def update_channels(stub): db_channel.auto_fees = pending_channel.auto_fees pending_channel.delete() if old_fee_rate is not None and old_fee_rate != local_policy.fee_rate_milli_msat: - print(f"{datetime.now().strftime('%c')} : [Data] : Ext fee change detected on {db_channel.chan_id} for peer {db_channel.alias}: fee updated from {old_fee_rate} to {db_channel.local_fee_rate}") - #External Fee change detected, update auto fee log + logger.info(f'Ext fee change detected on {db_channel.chan_id} for peer {db_channel.alias}: fee updated from {old_fee_rate} to {db_channel.local_fee_rate}') + # External Fee change detected, update auto fee log db_channel.fees_updated = datetime.now() - Autofees(chan_id=db_channel.chan_id, peer_alias=db_channel.alias, setting=(f"Ext"), old_value=old_fee_rate, new_value=db_channel.local_fee_rate).save() + Autofees(chan_id=db_channel.chan_id, peer_alias=db_channel.alias, setting=('Ext'), old_value=old_fee_rate, new_value=db_channel.local_fee_rate).save() db_channel.save() counter += 1 chan_list.append(channel.chan_id) records = Channels.objects.filter(is_open=True).count() if records > counter: - #A channel must have been closed, mark it as closed + # A channel must have been closed, mark it as closed channels = Channels.objects.filter(is_open=True).exclude(chan_id__in=chan_list) for channel in channels: channel.last_update = datetime.now() @@ -599,7 +600,7 @@ def get_tx_fees(txid): request_data = get(base_url + txid).json() fee = request_data['fee'] except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Data] : Error getting closure fees for {txid}: {str(e)}") + logger.error(f'Error getting closure fees for {txid}: {str(e)}') fee = 0 return fee @@ -618,7 +619,7 @@ def update_closures(stub): try: db_closure.save() except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Data] : Error inserting closure: {str(e)}") + logger.error(f'Error inserting closure: {str(e)}') Closures.objects.filter(funding_txid=txid,funding_index=index).delete() return if resolution_count > 0: @@ -638,19 +639,19 @@ def reconnect_peers(stub): if peers.filter(pubkey=inactive_peer).exists(): peer = peers.filter(pubkey=inactive_peer)[0] if peer.last_reconnected == None or (int((datetime.now() - peer.last_reconnected).total_seconds() / 60) > 2): - print(f"{datetime.now().strftime('%c')} : [Data] : Reconnecting peer {peer.alias} {peer.pubkey}, last reconnected at {peer.last_reconnected}") + logger.info(f'Reconnecting peer {peer.alias} {peer.pubkey}, last reconnected at {peer.last_reconnected}') if peer.connected == True: - print(f"{datetime.now().strftime('%c')} : [Data] : Inactive channel is still connected to peer, disconnecting peer {peer.alias} {inactive_peer}") + logger.info(f'Inactive channel is still connected to peer, disconnecting peer {peer.alias} {inactive_peer}') disconnectpeer(stub, peer) try: node = stub.GetNodeInfo(ln.NodeInfoRequest(pub_key=inactive_peer, include_channels=False)).node host = node.addresses[0].addr except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Data] : Unable to find node info on graph, using last known value for {peer.alias} {peer.pubkey} at {peer.address}: {str(e)}") + logger.error(f'Unable to find node info on graph, using last known value for {peer.alias} {peer.pubkey} at {peer.address}: {str(e)}') host = peer.address - print(f"{datetime.now().strftime('%c')} : [Data] : Attempting connection to {peer.alias} {inactive_peer} at {host}") + logger.info(f'Attempting connection to {peer.alias} {inactive_peer} at {host}') try: - #try both the graph value and last know value + # try both the graph value and last know value stub.ConnectPeer(request = ln.ConnectPeerRequest(addr=ln.LightningAddress(pubkey=inactive_peer, host=host), perm=True, timeout=5)) if host != peer.address and peer.address[:9] != '127.0.0.1': stub.ConnectPeer(request = ln.ConnectPeerRequest(addr=ln.LightningAddress(pubkey=inactive_peer, host=peer.address), perm=True, timeout=5)) @@ -659,7 +660,7 @@ def reconnect_peers(stub): details_index = error.find('details =') + 11 debug_error_index = error.find('debug_error_string =') - 3 error_msg = error[details_index:debug_error_index] - print(f"{datetime.now().strftime('%c')} : [Data] : Error reconnecting {peer.alias} {inactive_peer}: {error_msg}") + logger.error(f'Error reconnecting {peer.alias} {inactive_peer}: {error_msg}') peer.last_reconnected = datetime.now() peer.save() @@ -687,7 +688,7 @@ def clean_payments(stub): details_index = error.find('details =') + 11 debug_error_index = error.find('debug_error_string =') - 3 error_msg = error[details_index:debug_error_index] - print(f"{datetime.now().strftime('%c')} : [Data] : Error cleaning payment {payment.payment_hash} at index {payment.index} with payment status {payment.status}: {error_msg}") + logger.error(f'Error cleaning payment {payment.payment_hash} at index {payment.index} with payment status {payment.status}: {error_msg}') finally: payment.cleaned = True payment.save() @@ -727,19 +728,19 @@ def auto_fees(stub): inbound_base_fee = -channel.local_base_fee stub.UpdateChannelPolicy(ln.PolicyUpdateRequest(chan_point=channel_point, base_fee_msat=channel.local_base_fee, fee_rate=(target_channel['new_rate']/1000000), time_lock_delta=channel.local_cltv, inbound_fee=ln.InboundFee(base_fee_msat=inbound_base_fee, fee_rate_ppm=inbound_fee_rate))) if target_channel['inbound_adjustment'] != 0: - print(f"{datetime.now().strftime('%c')} : [Data] : Updating inbound fees for channel {str(target_channel['chan_id'])} to a value of: {str(target_channel['new_inbound_rate'])}") + logger.info(f'Updating inbound fees for channel {str(target_channel["chan_id"])} to a value of: {str(target_channel["new_inbound_rate"])}') channel.local_inbound_fee_rate = target_channel['new_inbound_rate'] InboundFeeLog(chan_id=channel.chan_id, peer_alias=channel.alias, setting=(f"AF [ {target_channel['net_routed_7day']}:{target_channel['in_percent']}:{target_channel['out_percent']} ]"), old_value=target_channel['local_inbound_fee_rate'], new_value=target_channel['new_inbound_rate']).save() else: stub.UpdateChannelPolicy(ln.PolicyUpdateRequest(chan_point=channel_point, base_fee_msat=channel.local_base_fee, fee_rate=(target_channel['new_rate']/1000000), time_lock_delta=channel.local_cltv)) if target_channel['adjustment'] != 0: - print(f"{datetime.now().strftime('%c')} : [Data] : Updating outbound fees for channel {str(target_channel['chan_id'])} to a value of: {str(target_channel['new_rate'])}") + logger.info(f'Updating outbound fees for channel {str(target_channel["chan_id"])} to a value of: {str(target_channel["new_rate"])}') channel.local_fee_rate = target_channel['new_rate'] Autofees(chan_id=channel.chan_id, peer_alias=channel.alias, setting=(f"AF [ {target_channel['net_routed_7day']}:{target_channel['in_percent']}:{target_channel['out_percent']} ]"), old_value=target_channel['local_fee_rate'], new_value=target_channel['new_rate']).save() channel.fees_updated = datetime.now() channel.save() except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Data] : Error processing auto_fees: {str(e)}") + logger.error(f'Error processing auto_fees: {str(e)}') def agg_htlcs(target_htlcs, category): @@ -770,7 +771,7 @@ def agg_htlcs(target_htlcs, category): htlc_itm.save() FailedHTLCs.objects.filter(id__in=target_ids, chan_id_in=htlc['chan_id_in'], chan_id_out=htlc['chan_id_out']).annotate(day=TruncDay('timestamp')).filter(day=htlc['day']).delete() except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Data] : Error processing agg_htlcs: {str(e)}") + logger.error(f'Error processing agg_htlcs: {str(e)}') def agg_failed_htlcs(): time_filter = datetime.now() - timedelta(days=30) @@ -780,10 +781,10 @@ def agg_failed_htlcs(): def main(): while True: - print(f"{datetime.now().strftime('%c')} : [Data] : Starting data execution...") + logger.info('Starting data execution...') try: stub = lnrpc.LightningStub(lnd_connect()) - #Update data + # Update data update_peers(stub) update_channels(stub) update_invoices(stub) @@ -796,8 +797,8 @@ def main(): auto_fees(stub) agg_failed_htlcs() except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Data] : Error processing background data: {str(e)}") - print(f"{datetime.now().strftime('%c')} : [Data] : Data execution completed...sleeping for 20 seconds") + logger.error(f'Error processing background data: {str(e)}') + logger.info('Data execution completed...sleeping for 20 seconds') sleep(20) if __name__ == '__main__': diff --git a/p2p.py b/p2p.py index e09abe5..1055d3d 100644 --- a/p2p.py +++ b/p2p.py @@ -1,5 +1,4 @@ import django, multiprocessing -from datetime import datetime from gui.lnd_deps import lightning_pb2_grpc as lnrpc from gui.lnd_deps.lnd_connect import lnd_connect from os import environ @@ -8,6 +7,8 @@ environ['DJANGO_SETTINGS_MODULE'] = 'lndg.settings' django.setup() from gui.models import LocalSettings from trade import serve_trades +import logging +logger = logging.getLogger('[P2P]') def trade(): stub = lnrpc.LightningStub(lnd_connect()) @@ -28,21 +29,21 @@ def main(): db_value = check_setting() if current_value != db_value: if db_value == 1: - print(f"{datetime.now().strftime('%c')} : [P2P] : Starting the p2p service...") + logger.info('Starting the p2p service...') django.db.connections.close_all() p2p_thread = multiprocessing.Process(target=trade) p2p_thread.start() else: if 'p2p_thread' in locals() and p2p_thread.is_alive(): - print(f"{datetime.now().strftime('%c')} : [P2P] : Stopping the p2p service...") + logger.info('Stopping the p2p service...') p2p_thread.terminate() current_value = db_value sleep(2) # polling interval except Exception as e: - print(f"{datetime.now().strftime('%c')} : [P2P] : Error running p2p service: {str(e)}") + logger.error(f'Error running p2p service: {str(e)}') finally: if 'p2p_thread' in locals() and p2p_thread.is_alive(): - print(f"{datetime.now().strftime('%c')} : [P2P] : Removing any remaining processes...") + logger.info('Removing any remaining processes...') p2p_thread.terminate() sleep(20) diff --git a/rebalancer.py b/rebalancer.py index d214dc9..60ccc55 100644 --- a/rebalancer.py +++ b/rebalancer.py @@ -10,31 +10,32 @@ from gui.lnd_deps import router_pb2_grpc as lnrouter from gui.lnd_deps.lnd_connect import lnd_connect, async_lnd_connect from os import environ from typing import List - environ['DJANGO_SETTINGS_MODULE'] = 'lndg.settings' django.setup() from gui.models import Rebalancer, Channels, LocalSettings, Forwards, Autopilot +import logging +logger = logging.getLogger('[Rebalancer]') @sync_to_async def get_out_cans(rebalance, auto_rebalance_channels): try: return list(auto_rebalance_channels.filter(auto_rebalance=False, percent_outbound__gte=F('ar_out_target')).exclude(remote_pubkey=rebalance.last_hop_pubkey).values_list('chan_id', flat=True)) except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error getting outbound cands: {str(e)}") + logger.error(f'Error getting outbound cands: {str(e)}') @sync_to_async def save_record(record): try: record.save() except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error saving database record: {str(e)}") + logger.error(f'Error saving database record: {str(e)}') @sync_to_async def inbound_cans_len(inbound_cans): try: return len(inbound_cans) except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error getting inbound cands: {str(e)}") + logger.error(f'Error getting inbound cands: {str(e)}') @sync_to_async def check_and_set_allow_multishards(): @@ -59,7 +60,7 @@ async def run_rebalancer(rebalance, worker): auto_rebalance_channels = Channels.objects.filter(is_active=True, is_open=True, private=False).annotate(percent_outbound=((Sum('local_balance')+Sum('pending_outbound')-rebalance.value)*100)/Sum('capacity')).annotate(inbound_can=(((Sum('remote_balance')+Sum('pending_inbound'))*100)/Sum('capacity'))/Sum('ar_in_target')) outbound_cans = await get_out_cans(rebalance, auto_rebalance_channels) if len(outbound_cans) == 0 and rebalance.manual == False: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : No outbound_cans") + logger.info('No outbound_cans') rebalance.status = 406 rebalance.start = datetime.now() rebalance.stop = datetime.now() @@ -75,7 +76,7 @@ async def run_rebalancer(rebalance, worker): chan_ids = json.loads(rebalance.outgoing_chan_ids) timeout = rebalance.duration * 60 invoice_response = stub.AddInvoice(ln.Invoice(value=rebalance.value, expiry=timeout)) - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : {worker} starting rebalance for {rebalance.target_alias} {rebalance.last_hop_pubkey} for {rebalance.value} sats and duration {rebalance.duration}, using {len(chan_ids)} outbound channels") + logger.debug(f'{worker} starting rebalance for {rebalance.target_alias} {rebalance.last_hop_pubkey} for {rebalance.value} sats and duration {rebalance.duration}, using {len(chan_ids)} outbound channels') async for payment_response in routerstub.SendPaymentV2(lnr.SendPaymentRequest(payment_request=str(invoice_response.payment_request), fee_limit_msat=int(rebalance.fee_limit*1000), outgoing_chan_ids=chan_ids, last_hop_pubkey=bytes.fromhex(rebalance.last_hop_pubkey), timeout_seconds=(timeout-5), allow_self_payment=True, max_parts=max_parts), timeout=(timeout+60)): if payment_response.status == 1 and rebalance.status == 0: #IN-FLIGHT @@ -111,11 +112,11 @@ async def run_rebalancer(rebalance, worker): rebalance.status = 408 else: rebalance.status = 400 - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error while sending payment: {str(e)}") + logger.error(f'Error while sending payment: {str(e)}') finally: rebalance.stop = datetime.now() await save_record(rebalance) - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : {worker} completed payment attempts for: {rebalance.payment_hash}") + logger.debug(f'{worker} completed payment attempts for: {rebalance.payment_hash}') original_alias = rebalance.target_alias inc=1.21 dec=2 @@ -128,7 +129,7 @@ async def run_rebalancer(rebalance, worker): if await inbound_cans_len(inbound_cans) > 0 and len(outbound_cans) > 0: next_rebalance = Rebalancer(value=int(rebalance.value*inc), fee_limit=round(rebalance.fee_limit*inc, 3), outgoing_chan_ids=str(outbound_cans).replace('\'', ''), last_hop_pubkey=rebalance.last_hop_pubkey, target_alias=original_alias, duration=1) await save_record(next_rebalance) - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : RapidFire increase for {next_rebalance.target_alias} from {rebalance.value} to {next_rebalance.value}") + logger.info(f'RapidFire increase for {next_rebalance.target_alias} from {rebalance.value} to {next_rebalance.value}') else: next_rebalance = None # For failed rebalances, try in rapid fire with reduced balances until give up. @@ -146,14 +147,14 @@ async def run_rebalancer(rebalance, worker): if await inbound_cans_len(inbound_cans) > 0 and len(outbound_cans) > 0: next_rebalance = Rebalancer(value=int(next_value), fee_limit=round(rebalance.fee_limit/(rebalance.value/next_value), 3), outgoing_chan_ids=str(outbound_cans).replace('\'', ''), last_hop_pubkey=rebalance.last_hop_pubkey, target_alias=original_alias, duration=1) await save_record(next_rebalance) - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : RapidFire decrease for {next_rebalance.target_alias} from {rebalance.value} to {next_rebalance.value}") + logger.info(f'RapidFire decrease for {next_rebalance.target_alias} from {rebalance.value} to {next_rebalance.value}') else: next_rebalance = None else: next_rebalance = None return next_rebalance except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error running rebalance attempt: {str(e)}") + logger.error(f'Error running rebalance attempt: {str(e)}') @sync_to_async def estimate_liquidity( payment ): @@ -166,9 +167,9 @@ def estimate_liquidity( payment ): if attempt.failure.failure_source_index == total_hops: #Failure from last hop indicating liquidity available estimated_liquidity = attempt.route.total_amt if attempt.route.total_amt > estimated_liquidity else estimated_liquidity - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Estimated Liquidity {estimated_liquidity} for payment {payment.payment_hash} with status {payment.status} and reason {payment.failure_reason}") + logger.info(f'Estimated Liquidity {estimated_liquidity} for payment {payment.payment_hash} with status {payment.status} and reason {payment.failure_reason}') except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error estimating liquidity: {str(e)}") + logger.error(f'Error estimating liquidity: {str(e)}') estimated_liquidity = 0 return estimated_liquidity @@ -189,7 +190,7 @@ def update_channels(stub, incoming_channel, outgoing_channel): db_channel.remote_balance = channel.remote_balance db_channel.save() except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error updating channel balances: {str(e)}") + logger.error(f'Error updating channel balances: {str(e)}') @sync_to_async def auto_schedule() -> List[Rebalancer]: @@ -255,15 +256,15 @@ def auto_schedule() -> List[Rebalancer]: last_rebalance = Rebalancer.objects.filter(last_hop_pubkey=target.remote_pubkey).exclude(status=0).order_by('-id')[0] if not (last_rebalance.status == 2 or (last_rebalance.status > 2 and (int((datetime.now() - last_rebalance.stop).total_seconds() / 60) > wait_period)) or (last_rebalance.status == 1 and ((int((datetime.now() - last_rebalance.start).total_seconds() / 60) - last_rebalance.duration) > wait_period))): continue - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Creating Auto Rebalance Request for: {target.chan_id}") - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Value: {target_value} / {target.ar_amt_target} | Fee: {target_fee} | Duration: {target_time}") - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Request routing outbound via: {outbound_cans}") + logger.info(f'Creating Auto Rebalance Request for: {target.chan_id}') + logger.info(f'Value: {target_value} / {target.ar_amt_target} | Fee: {target_fee} | Duration: {target_time}') + logger.info(f'Request routing outbound via: {outbound_cans}') new_rebalance = Rebalancer(value=target_value, fee_limit=target_fee, outgoing_chan_ids=str(outbound_cans).replace('\'', ''), last_hop_pubkey=target.remote_pubkey, target_alias=target.alias, duration=target_time) new_rebalance.save() to_schedule.append(new_rebalance) return to_schedule except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error scheduling rebalances: {str(e)}") + logger.error(f'Error scheduling rebalances: {str(e)}') return to_schedule @sync_to_async @@ -294,32 +295,28 @@ def auto_enable(): oapD = 0 if routed_out_apday == 0 else int(forwards.filter(chan_id_out__in=chan_list).aggregate(Sum('amt_out_msat'))['amt_out_msat__sum']/10000000)/100 for peer_channel in lookup_channels.filter(chan_id__in=chan_list): if peer_channel.ar_out_target == 100 and peer_channel.auto_rebalance == True: - #Special Case for LOOP, Wos, etc. Always Auto Rebalance if enabled to keep outbound full. - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Skipping AR enabled and 100% oTarget channel: {peer_channel.alias} {peer_channel.chan_id}") - pass + logger.debug('Special case for sinks like LOOP, Wos, etc. if AR enabled and oTarget at 100%: Pass') + logger.info(f'Skipping AR enabled and 100% oTarget channel: {peer_channel.alias} {peer_channel.chan_id}') elif oapD > (iapD*1.10) and outbound_percent > 75: - #print('Case 1: Pass') - pass + logger.debug('Auto-Enable Case 1: Pass') elif oapD > (iapD*1.10) and inbound_percent > 75 and peer_channel.auto_rebalance == False: - #print('Case 2: Enable AR - o7D > i7D AND Inbound Liq > 75%') + logger.debug('Case 2: Enable AR - o7D > i7D AND Inbound Liq > 75%') peer_channel.auto_rebalance = True peer_channel.save() Autopilot(chan_id=peer_channel.chan_id, peer_alias=peer_channel.alias, setting='Enabled', old_value=0, new_value=1).save() - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Auto Pilot Enabled for {peer_channel.alias} {peer_channel.chan_id}: {oapD} {iapD}") + logger.info(f'Auto Pilot Enabled for {peer_channel.alias} {peer_channel.chan_id}: {oapD} {iapD}') elif oapD < (iapD*1.10) and outbound_percent > 75 and peer_channel.auto_rebalance == True: - #print('Case 3: Disable AR - o7D < i7D AND Outbound Liq > 75%') + logger.debug('Case 3: Disable AR - o7D < i7D AND Outbound Liq > 75%') peer_channel.auto_rebalance = False peer_channel.save() Autopilot(chan_id=peer_channel.chan_id, peer_alias=peer_channel.alias, setting='Enabled', old_value=1, new_value=0).save() - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Auto Pilot Disabled for {peer_channel.alias} {peer_channel.chan_id}: {oapD} {iapD}" ) + logger.info(f'Auto Pilot Disabled for {peer_channel.alias} {peer_channel.chan_id}: {oapD} {iapD}') elif oapD < (iapD*1.10) and inbound_percent > 75: - #print('Case 4: Pass') - pass + logger.debug('Case 4: Pass') else: - #print('Case 5: Pass') - pass + logger.debug('Case 5: Pass') except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error during auto channel enabling: {str(e)}") + logger.error(f'Error during auto channel enabling: {str(e)}') @sync_to_async def get_pending_rebals(): @@ -327,49 +324,49 @@ def get_pending_rebals(): rebalances = Rebalancer.objects.filter(status=0).order_by('id') return rebalances, len(rebalances) except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Error getting pending rebalances: {str(e)}") + logger.error(f'Error getting pending rebalances: {str(e)}') async def async_queue_manager(rebalancer_queue): global scheduled_rebalances, active_rebalances, shutdown_rebalancer - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue manager is starting...") + logger.debug('Queue manager is starting...') try: while True: if shutdown_rebalancer == True: return - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue currently has {rebalancer_queue.qsize()} items...") - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : There are currently {len(active_rebalances)} tasks in progress...") - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue manager is checking for more work...") + logger.info(f'Queue currently has {rebalancer_queue.qsize()} items...') if rebalancer_queue.qsize() > 0 else logger.debug('Queue currently has no items...') + logger.info(f'There are currently {len(active_rebalances)} tasks in progress...') if len(active_rebalances) > 0 else logger.debug('There are currently no tasks in progress...') + logger.info('Queue manager is checking for more work...') pending_rebalances, rebal_count = await get_pending_rebals() if rebal_count > 0: for rebalance in pending_rebalances: if rebalance.id not in (scheduled_rebalances + active_rebalances): - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Found a pending job to schedule with id: {rebalance.id}") + logger.info(f'Found a pending job to schedule with id: {rebalance.id}') scheduled_rebalances.append(rebalance.id) await rebalancer_queue.put(rebalance) await auto_enable() scheduled = await auto_schedule() if len(scheduled) > 0: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Scheduling {len(scheduled)} more jobs...") + logger.info(f'Scheduling {len(scheduled)} more jobs...') for rebalance in scheduled: scheduled_rebalances.append(rebalance.id) await rebalancer_queue.put(rebalance) elif rebalancer_queue.qsize() == 0 and len(active_rebalances) == 0: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue is still empty, stopping the rebalancer...") + logger.info('No active work found, stopping the rebalancer...') shutdown_rebalancer = True return await asyncio.sleep(30) except Exception as e: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue manager exception: {str(e)}") + logger.error(f'Queue manager exception: {str(e)}') shutdown_rebalancer = True finally: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Queue manager has shut down...") + logger.debug('Queue manager has shut down...') async def async_run_rebalancer(worker, rebalancer_queue): global scheduled_rebalances, active_rebalances, shutdown_rebalancer while True: if not rebalancer_queue.empty() and not shutdown_rebalancer: rebalance = await rebalancer_queue.get() - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : {worker} is starting a new request...") + logger.debug(f'{worker} is starting a new request...') active_rebalance_id = None if rebalance != None: active_rebalance_id = rebalance.id @@ -379,7 +376,7 @@ async def async_run_rebalancer(worker, rebalancer_queue): rebalance = await run_rebalancer(rebalance, worker) if active_rebalance_id != None: active_rebalances.remove(active_rebalance_id) - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : {worker} completed its request...") + logger.debug(f'{worker} completed its request...') else: if shutdown_rebalancer == True: return @@ -390,7 +387,7 @@ async def start_queue(worker_count=1): manager = asyncio.create_task(async_queue_manager(rebalancer_queue)) workers = [asyncio.create_task(async_run_rebalancer("Worker " + str(worker_num+1), rebalancer_queue)) for worker_num in range(worker_count)] await asyncio.gather(manager, *workers) - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Manager and workers have stopped...") + logger.debug('Manager and workers have stopped...') @sync_to_async def get_worker_count(): @@ -406,7 +403,7 @@ async def update_worker_count(): if updated_worker_count != worker_count: worker_count = updated_worker_count shutdown_rebalancer = True - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : New worker count detected...restarting rebalancer") + logger.info('New worker count detected...restarting rebalancer') await asyncio.sleep(20) def main(): @@ -417,7 +414,7 @@ def main(): LocalSettings(key='AR-Workers', value='1').save() worker_count = 1 try: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Rebalancer initializing...") + logger.info('Rebalancer initializing...') loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.create_task(update_worker_count()) @@ -432,13 +429,13 @@ def main(): unknown_error.stop = datetime.now() unknown_error.save() loop.run_until_complete(start_queue(worker_count)) - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Rebalancer successfully exited...sleeping for 20 seconds") + logger.info('Rebalancer successfully exited...sleeping for 20 seconds') sleep(20) except Exception as e: error = str(e) - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Rebalancer loop error: {error}") + logger.error(f'Rebalancer loop error: {error}') finally: - print(f"{datetime.now().strftime('%c')} : [Rebalancer] : Rebalancer loop has been terminated") + logger.info('Rebalancer loop has been terminated') if __name__ == '__main__': main() diff --git a/trade.py b/trade.py index 5ee6475..a79df6b 100644 --- a/trade.py +++ b/trade.py @@ -19,6 +19,8 @@ from os import environ environ['DJANGO_SETTINGS_MODULE'] = 'lndg.settings' django.setup() from gui.models import TradeSales, Payments, PaymentHops, Forwards, Peers +import logging +logger = logging.getLogger('[P2P]') def is_hex(n): return len(n) % 2 == 0 and all(c in '0123456789ABCDEFabcdef' for c in n) @@ -667,7 +669,7 @@ def getSecret(stub, sale_type): outgoing_nodes = Forwards.objects.filter(forward_date__gte=filter_30day).values('chan_id_out').annotate(ppm=Round((Sum('fee')/Sum('amt_out_msat'))*1000000000, output_field=IntegerField()), score=Round((Round(Count('id')/1, output_field=IntegerField())+Round(Sum('amt_out_msat')/100000, output_field=IntegerField()))/10, output_field=IntegerField())).exclude(score=0).order_by('-score', '-ppm')[:5] secret = json.dumps({"incoming_nodes":list(incoming_nodes.values('chan_id_in', 'score', 'ppm')), "outgoing_nodes":list(outgoing_nodes.values('chan_id_out', 'score', 'ppm'))}) except Exception as e: - print(f"{datetime.now().strftime('%c')} : [P2P] : Error getting secret: {str(e)}") + logger.error(f'Error getting secret: {str(e)}') secret = None finally: return secret @@ -680,7 +682,7 @@ def getSecret(stub, sale_type): payment_nodes = PaymentHops.objects.filter(payment_hash__in=payments_30day).exclude(node_pubkey=self_pubkey).values('node_pubkey').annotate(ppm=Round((Sum('fee')/Sum('amt'))*1000000, output_field=IntegerField()), score=Round((Round(Count('id')/1, output_field=IntegerField())+Round(Sum('amt')/100000, output_field=IntegerField()))/10, output_field=IntegerField())).exclude(score=0).order_by('-score', 'ppm')[:10] secret = json.dumps({"payment_nodes": list(payment_nodes.values('node_pubkey', 'score', 'ppm'))}) except Exception as e: - print(f"{datetime.now().strftime('%c')} : [P2P] : Error getting secret: {str(e)}") + logger.error(f'Error getting secret: {str(e)}') secret = None finally: return secret @@ -688,9 +690,9 @@ def getSecret(stub, sale_type): return None def serve_trades(stub): - print(f"{datetime.now().strftime('%c')} : [P2P] : Serving trades...") + logger.info('Serving trades...') for trade in get_trades(): - print(f"{datetime.now().strftime('%c')} : [P2P] : Serving trade: {trade.id}") + logger.info(f'Serving trade: {trade.id}') for response in stub.SubscribeCustomMessages(ln.SubscribeCustomMessagesRequest()): if response.type == 32768: from_peer = response.peer @@ -703,7 +705,7 @@ def serve_trades(stub): if 'type' in request: req_type = request['type'] if req_type == '8050005': # request a seller to finalize a trade or give all open trades - print(f"{datetime.now().strftime('%c')} : [P2P] : SELLER ACTION", '|', 'ID:', request['id'], '|', 'Records:', request['records']) + logger.info(f'SELLER ACTION | ID: {request["id"]} | Records: {request["records"]}') select_trade = next((record for record in request['records'] if record['type'] == '0'), None) request_trade = next((record for record in request['records'] if record['type'] == '1'), None) if request_trade: @@ -723,7 +725,7 @@ def serve_trades(stub): else: secret = trade_details.secret if not secret: - print(f"{datetime.now().strftime('%c')} : [P2P] : Failed to get secret for:", trade_details.id) + logger.error(f'Failed to get secret for: {trade_details.id}') continue signerstub = lnsigner.SignerStub(lnd_connect()) shared_key = signerstub.DeriveSharedKey(lns.SharedKeyRequest(ephemeral_pubkey=from_peer)).shared_key @@ -743,16 +745,16 @@ def serve_trades(stub): trade_details.save() stub.SendCustomMessage(ln.SendCustomMessageRequest(peer=from_peer, type=32768, data=bytes.fromhex(trade_data))) else: - print(f"{datetime.now().strftime('%c')} : [P2P] : Expected request type in message:", request['id']) + logger.error(f'Expected request type in message: {request["id"]}') if 'response' in msg_response: request = msg_response['response'] if 'failure' in request and request['failure'] != None: # failure message returned - print(f"{datetime.now().strftime('%c')} : [P2P] : Failure:", request['failure']) + logger.error(f'Failure: {request["failure"]}') else: if len(request['records']) == 0: # message acknowledgements - print(f"{datetime.now().strftime('%c')} : [P2P] : ACK", '|', 'ID:', request['id']) + logger.info(f'ACK | ID: {request["id"]}') async def get_open_trades(astub, results): try: