mirror of
https://github.com/cryptosharks131/lndg.git
synced 2026-08-13 12:33:04 +02:00
Merge branch 'v1.11.0' of github.com:cryptosharks131/lndg into v1.11.0
This commit is contained in:
cryptosharks131
commit
6b9bfe5cff
8 changed files with 194 additions and 122 deletions
|
|
@ -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()
|
||||
|
|
|
|||
29
gui/views.py
29
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'])
|
||||
|
|
|
|||
|
|
@ -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__':
|
||||
|
|
|
|||
|
|
@ -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...')
|
||||
|
|
|
|||
71
jobs.py
71
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__':
|
||||
|
|
|
|||
11
p2p.py
11
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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
20
trade.py
20
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:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue