110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Crypto Safety Check — verify current positions against 30% utilization cap.
|
|
|
|
Usage:
|
|
python3 safety_check.py [--symbol SYMBOL] [--market-cap 30]
|
|
|
|
Reads OKX credentials from ~/.bashrc, queries swap positions, and reports:
|
|
- Total margin / free balance ratio (utilization %)
|
|
- Per-symbol: margin, contracts, direction, leverage, liq price
|
|
- Verdict: SAFE / OVER-CAP / NO-POSITION
|
|
|
|
Does NOT place any orders. Read-only diagnostic.
|
|
|
|
The 30% cap is the user's explicit safety rule (2026-07-08), overriding the
|
|
default advisor script value of 45% in config.json.
|
|
"""
|
|
import argparse
|
|
import os
|
|
import re
|
|
import sys
|
|
import ccxt
|
|
|
|
# Load OKX creds from bashrc (avoid source; bashrc has non-interactive guard)
|
|
def load_creds():
|
|
creds = {}
|
|
with open(os.path.expanduser('~/.bashrc')) as f:
|
|
for line in f:
|
|
m = re.match(r'export\s+(OKX_\w+)=(.*)', line.strip())
|
|
if m and '...' not in m.group(2):
|
|
creds[m.group(1)] = m.group(2).strip().strip('"').strip("'")
|
|
return creds
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--symbol', help='Filter to single symbol (e.g. ETH)')
|
|
parser.add_argument('--market-cap', type=float, default=40.0,
|
|
help='Safety utilization %% (default 40)')
|
|
args = parser.parse_args()
|
|
|
|
creds = load_creds()
|
|
if not all(k in creds for k in ['OKX_API_KEY', 'OKX_SECRET', 'OKX_PASSPHRASE']):
|
|
print('ERROR: OKX credentials missing in ~/.bashrc', file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
ex = ccxt.okx({
|
|
'apiKey': creds['OKX_API_KEY'],
|
|
'secret': creds['OKX_SECRET'],
|
|
'password': creds['OKX_PASSPHRASE'],
|
|
'proxies': {'http': 'http://127.0.0.1:7890',
|
|
'https': 'http://127.0.0.1:7890'},
|
|
'timeout': 30000,
|
|
})
|
|
ex.options['defaultType'] = 'swap'
|
|
|
|
# Query positions
|
|
positions = ex.fetch_positions()
|
|
active = [p for p in positions if abs(float(p.get('contracts', 0))) > 0]
|
|
|
|
if args.symbol:
|
|
active = [p for p in active if args.symbol.upper() in p['symbol'].upper()]
|
|
|
|
# Query balance
|
|
bal = ex.fetch_balance()
|
|
free = float(bal.get('USDT', {}).get('free', 0))
|
|
total_eq = float(bal.get('USDT', {}).get('total', 0))
|
|
|
|
# Compute total margin
|
|
total_margin = 0.0
|
|
print(f'\n=== {args.symbol or "ALL"} Positions ===')
|
|
print(f'{"Symbol":<12} {"Side":<6} {"Qty":<8} {"Entry":<10} {"Mark":<10} '
|
|
f'{"Margin":<10} {"Lever":<6} {"UPL":<10}')
|
|
print('-' * 80)
|
|
for p in active:
|
|
sym = p['symbol']
|
|
contracts = float(p['contracts'])
|
|
side = 'long' if contracts > 0 else 'short'
|
|
entry = float(p.get('entryPrice', 0))
|
|
mark = float(p.get('markPrice', 0))
|
|
margin = float(p.get('initialMargin', 0))
|
|
lever = p.get('leverage', '?')
|
|
upl = float(p.get('unrealizedPnl', 0))
|
|
total_margin += margin
|
|
print(f'{sym:<12} {side:<6} {contracts:<8.2f} {entry:<10.2f} '
|
|
f'{mark:<10.2f} {margin:<10.2f} {str(lever):<6} {upl:<+10.2f}')
|
|
|
|
print('-' * 80)
|
|
util = (total_margin / free * 100) if free > 0 else 999.0
|
|
print(f'\nTotal margin: {total_margin:.2f} USDT')
|
|
print(f'Free balance: {free:.2f} USDT')
|
|
print(f'Total equity: {total_eq:.2f} USDT')
|
|
print(f'Utilization: {util:.1f}% (cap: {args.market_cap:.0f}%)')
|
|
|
|
if util > args.market_cap:
|
|
over_by = total_margin - (free * args.market_cap / 100)
|
|
print(f'\n⚠️ OVER SAFETY CAP by {over_by:.2f} USDT')
|
|
print(f' Reduce positions or top up balance.')
|
|
sys.exit(2)
|
|
elif not active:
|
|
print('\n✅ No active positions.')
|
|
sys.exit(0)
|
|
else:
|
|
headroom = free * args.market_cap / 100 - total_margin
|
|
print(f'\n✅ Within safety cap. Headroom: {headroom:.2f} USDT')
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main() |