468 lines
14 KiB
Python
Executable file
468 lines
14 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
# AUTHOR: Daesorin
|
|
# CREATED: 2026-05-26 12:45 +0100
|
|
# UPDATED: 2026-07-12 15:13 +0100
|
|
"""
|
|
rate v0.6
|
|
terminal multi-asset valuation engine. runs entirely on the python standard
|
|
library with zero external dependencies. supports specialized workflows for
|
|
forex, precious metals, energy futures, market indices, crypto, equities, and ETFs.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import socket
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
# FORCE IPV4 RESOLUTION
|
|
# prevent unreachable network errors on broken ipv6 internet gateways
|
|
_old_getaddrinfo = socket.getaddrinfo
|
|
|
|
|
|
def _ipv4_getaddrinfo(*args, **kwargs):
|
|
responses = _old_getaddrinfo(*args, **kwargs)
|
|
return [res for res in responses if res[0] == socket.AF_INET]
|
|
|
|
|
|
socket.getaddrinfo = _ipv4_getaddrinfo
|
|
|
|
# TERMINAL COLOURS
|
|
|
|
|
|
def get_colours():
|
|
if not sys.stdout.isatty() or os.environ.get("NO_COLOR"):
|
|
return {
|
|
"red": "",
|
|
"green": "",
|
|
"yellow": "",
|
|
"blue": "",
|
|
"cyan": "",
|
|
"nc": "",
|
|
}
|
|
return {
|
|
"red": "\033[1;31m",
|
|
"green": "\033[1;32m",
|
|
"yellow": "\033[1;33m",
|
|
"blue": "\033[1;34m",
|
|
"cyan": "\033[1;36m",
|
|
"nc": "\033[0m",
|
|
}
|
|
|
|
|
|
COL = get_colours()
|
|
|
|
# ASSET MAPS
|
|
CRYPTO_MAP = {
|
|
"btc": "bitcoin",
|
|
"eth": "ethereum",
|
|
"usdt": "tether",
|
|
"bnb": "binancecoin",
|
|
"sol": "solana",
|
|
"xrp": "ripple",
|
|
"usdc": "usd-coin",
|
|
"ada": "cardano",
|
|
"doge": "dogecoin",
|
|
"ton": "the-open-network",
|
|
"trx": "tron",
|
|
"avax": "avalanche-2",
|
|
"shib": "shiba-inu",
|
|
"dot": "polkadot",
|
|
"link": "chainlink",
|
|
}
|
|
|
|
METAL_MAP = {
|
|
"gold": "GC=F",
|
|
"silver": "SI=F",
|
|
"platinum": "PL=F",
|
|
"palladium": "PA=F",
|
|
"copper": "HG=F",
|
|
}
|
|
|
|
ENERGY_MAP = {
|
|
"oil": "CL=F",
|
|
"wti": "CL=F",
|
|
"brent": "BZ=F",
|
|
"gas": "NG=F",
|
|
"natgas": "NG=F",
|
|
"heating": "HO=F",
|
|
}
|
|
|
|
INDEX_MAP = {
|
|
"sp500": "^GSPC",
|
|
"s&p500": "^GSPC",
|
|
"nasdaq": "^IXIC",
|
|
"dow": "^DJI",
|
|
"dji": "^DJI",
|
|
"ftse": "^FTSE",
|
|
"nikkei": "^N225",
|
|
"dax": "^GDAXI",
|
|
}
|
|
|
|
# FORMATTING UTILITIES
|
|
|
|
|
|
def fmt_num(val, decimals=2):
|
|
return f"{val:,.{decimals}f}"
|
|
|
|
|
|
def fmt_rate(val):
|
|
if val == 0:
|
|
return "0.00"
|
|
if abs(val) < 0.0001:
|
|
return f"{val:,.8f}".rstrip("0").rstrip(".")
|
|
if abs(val) < 1.0:
|
|
return f"{val:,.4f}"
|
|
return f"{val:,.2f}"
|
|
|
|
|
|
# CORE HTTP ENGINE
|
|
|
|
|
|
def fetch_json(url, params=None):
|
|
if params:
|
|
url = f"{url}?{urllib.parse.urlencode(params)}"
|
|
req = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
|
"Accept": "application/json",
|
|
},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as e:
|
|
print(
|
|
f"{COL['red']}error: api request failed with status {e.code}.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
except urllib.error.URLError:
|
|
print(
|
|
f"{COL['red']}error: network connection timed out or unreachable.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
except json.JSONDecodeError:
|
|
print(
|
|
f"{COL['red']}error: failed to parse remote json payload.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
|
|
def get_forex_rate(base, target):
|
|
base, target = base.upper(), target.upper()
|
|
if base == target:
|
|
return 1.0
|
|
data = fetch_json(f"https://open.er-api.com/v6/latest/{base}")
|
|
rates = data.get("rates", {})
|
|
if target not in rates:
|
|
print(
|
|
f"{COL['red']}error: currency target '{target}' unsupported.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
return float(rates[target])
|
|
|
|
|
|
def fetch_yahoo_market_data(ticker):
|
|
encoded = urllib.parse.quote(ticker)
|
|
url = f"https://query2.finance.yahoo.com/v8/finance/chart/{encoded}"
|
|
data = fetch_json(url, {"interval": "1d", "range": "1d"})
|
|
result = data.get("chart", {}).get("result")
|
|
if not result:
|
|
print(
|
|
f"{COL['red']}error: market asset '{ticker}' not found on yahoo finance.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
meta = result[0].get("meta", {})
|
|
price = float(meta.get("regularMarketPrice", 0.0))
|
|
prev_close = float(
|
|
meta.get("chartPreviousClose")
|
|
or meta.get("previousClose")
|
|
or price
|
|
)
|
|
return {
|
|
"symbol": meta.get("symbol", ticker.upper()),
|
|
"name": (
|
|
meta.get("shortName") or meta.get("longName") or ticker
|
|
).strip(),
|
|
"price": price,
|
|
"currency": meta.get("currency", "USD").upper(),
|
|
"change_pct": ((price - prev_close) / prev_close * 100.0) if prev_close else 0.0,
|
|
}
|
|
|
|
|
|
# UTILITY PARSER FOR MARKET INPUTS
|
|
|
|
|
|
def parse_market_args(args, default_target="NGN"):
|
|
amount = 1.0
|
|
target = default_target
|
|
if len(args) >= 2:
|
|
if re.match(r"^[0-9]+(\.[0-9]+)?$", args[1]):
|
|
amount = float(args[1])
|
|
if len(args) >= 3:
|
|
target = args[2]
|
|
else:
|
|
target = args[1]
|
|
if len(args) >= 3 and re.match(r"^[0-9]+(\.[0-9]+)?$", args[2]):
|
|
amount = float(args[2])
|
|
return amount, target.upper()
|
|
|
|
|
|
# SUBCOMMAND EXECUTORS
|
|
|
|
|
|
def cmd_forex(args):
|
|
amount, base, target = 1.0, "USD", "NGN"
|
|
if len(args) == 1:
|
|
if re.match(r"^[0-9]+(\.[0-9]+)?$", args[0]):
|
|
amount = float(args[0])
|
|
else:
|
|
base = args[0]
|
|
elif len(args) == 2:
|
|
if re.match(r"^[0-9]+(\.[0-9]+)?$", args[0]):
|
|
amount, base = float(args[0]), args[1]
|
|
else:
|
|
base, target = args[0], args[1]
|
|
elif len(args) >= 3:
|
|
if re.match(r"^[0-9]+(\.[0-9]+)?$", args[0]):
|
|
amount, base, target = float(args[0]), args[1], args[2]
|
|
else:
|
|
base, target, amount = args[0], args[1], float(args[2])
|
|
|
|
base, target = base.upper(), target.upper()
|
|
print(
|
|
f"{COL['blue']}fetching exchange rate for {base}/{target}...{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
rate = get_forex_rate(base, target)
|
|
val = amount * rate
|
|
print("-----------------------------------")
|
|
print(f"pair: {base} / {target}")
|
|
print(f"rate: {COL['yellow']}1 {base} = {fmt_rate(rate)} {target}{COL['nc']}")
|
|
print(
|
|
f"value: {COL['green']}{target} {fmt_num(val, 2)}{COL['nc']} ({base} {fmt_num(amount, 2)})"
|
|
)
|
|
print("-----------------------------------")
|
|
|
|
|
|
def cmd_yahoo_generic(ticker, amount, target_curr, class_label):
|
|
print(
|
|
f"{COL['blue']}fetching evaluation for {class_label} [{ticker}]...{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
market = fetch_yahoo_market_data(ticker)
|
|
native_curr = market["currency"]
|
|
native_price = market["price"]
|
|
total_native = amount * native_price
|
|
|
|
if native_curr != target_curr:
|
|
fx_rate = get_forex_rate(native_curr, target_curr)
|
|
final_val = total_native * fx_rate
|
|
rate_disp = f"{fmt_rate(native_price)} {native_curr} (~{fmt_num(native_price * fx_rate, 2)} {target_curr})"
|
|
val_disp = f"{COL['green']}{fmt_num(final_val, 2)} {target_curr}{COL['nc']} ({fmt_num(total_native, 2)} {native_curr})"
|
|
else:
|
|
rate_disp = f"{fmt_rate(native_price)} {native_curr}"
|
|
val_disp = f"{COL['green']}{fmt_num(total_native, 2)} {native_curr}{COL['nc']}"
|
|
|
|
chg = market["change_pct"]
|
|
chg_col = COL["green"] if chg >= 0 else COL["red"]
|
|
chg_sign = "+" if chg >= 0 else ""
|
|
|
|
print("-----------------------------------")
|
|
print(f"asset: {market['name']} ({market['symbol']})")
|
|
print(f"valuation: {COL['yellow']}{rate_disp}{COL['nc']}")
|
|
print(f"24h move: {chg_col}{chg_sign}{chg:.2f}%{COL['nc']}")
|
|
if amount != 1.0 or native_curr != target_curr:
|
|
print(f"total sum: {val_disp}")
|
|
print("-----------------------------------")
|
|
|
|
|
|
def cmd_metal(args):
|
|
if not args:
|
|
print(
|
|
f"{COL['red']}error: missing metal identifier.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
query = args[0].lower().strip()
|
|
ticker = METAL_MAP.get(query, args[0].upper())
|
|
amount, target = parse_market_args(args)
|
|
cmd_yahoo_generic(ticker, amount, target, "Precious Metal")
|
|
|
|
|
|
def cmd_energy(args):
|
|
if not args:
|
|
print(
|
|
f"{COL['red']}error: missing energy identifier.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
query = args[0].lower().strip()
|
|
ticker = ENERGY_MAP.get(query, args[0].upper())
|
|
amount, target = parse_market_args(args)
|
|
cmd_yahoo_generic(ticker, amount, target, "Energy Future")
|
|
|
|
|
|
def cmd_indices(args):
|
|
if not args:
|
|
print(
|
|
f"{COL['red']}error: missing index symbol.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
query = args[0].lower().strip()
|
|
ticker = INDEX_MAP.get(query, args[0].upper())
|
|
amount, target = parse_market_args(args)
|
|
cmd_yahoo_generic(ticker, amount, target, "Market Index")
|
|
|
|
|
|
def cmd_stock(args):
|
|
if not args:
|
|
print(
|
|
f"{COL['red']}error: missing stock ticker.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
ticker = args[0].upper()
|
|
amount, target = parse_market_args(args)
|
|
cmd_yahoo_generic(ticker, amount, target, "Equity Asset")
|
|
|
|
|
|
def cmd_etf(args):
|
|
if not args:
|
|
print(
|
|
f"{COL['red']}error: missing ETF symbol.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
ticker = args[0].upper()
|
|
amount, target = parse_market_args(args)
|
|
cmd_yahoo_generic(ticker, amount, target, "Exchange Traded Fund")
|
|
|
|
|
|
def cmd_crypto(args):
|
|
if not args:
|
|
print(
|
|
f"{COL['red']}error: missing crypto token name or symbol.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
query = args[0].lower().strip()
|
|
amount, vs_curr = parse_market_args(args, default_target="ngn")
|
|
vs_target = vs_curr.lower()
|
|
|
|
if query in CRYPTO_MAP:
|
|
coin_id, symbol = CRYPTO_MAP[query], query.upper()
|
|
else:
|
|
print(
|
|
f"{COL['blue']}searching coingecko for '{query}'...{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
search = fetch_json(
|
|
"https://api.coingecko.com/api/v3/search", {"query": query}
|
|
)
|
|
coins = search.get("coins", [])
|
|
if not coins:
|
|
print(
|
|
f"{COL['red']}error: token '{query}' not found.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
coin_id, symbol = coins[0]["id"], coins[0]["symbol"].upper()
|
|
|
|
print(
|
|
f"{COL['blue']}fetching {symbol} spot data in {vs_curr.upper()}...{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
data = fetch_json(
|
|
"https://api.coingecko.com/api/v3/simple/price",
|
|
{
|
|
"ids": coin_id,
|
|
"vs_currencies": vs_target,
|
|
"include_24hr_change": "true",
|
|
},
|
|
)
|
|
coin_data = data.get(coin_id, {})
|
|
if vs_target not in coin_data:
|
|
print(
|
|
f"{COL['red']}error: target market currency '{vs_curr}' unsupported.{COL['nc']}",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
rate = float(coin_data[vs_target])
|
|
change = float(coin_data.get(f"{vs_target}_24h_change", 0.0))
|
|
val = amount * rate
|
|
chg_col = COL["green"] if change >= 0 else COL["red"]
|
|
chg_sign = "+" if change >= 0 else ""
|
|
|
|
print("-----------------------------------")
|
|
print(f"crypto: {symbol} ({coin_id})")
|
|
print(f"spot rate: {COL['yellow']}{fmt_rate(rate)} {vs_curr.upper()}{COL['nc']}")
|
|
print(f"24h move: {chg_col}{chg_sign}{change:.2f}%{COL['nc']}")
|
|
if amount != 1.0:
|
|
print(
|
|
f"valuation: {COL['green']}{fmt_num(val, 2)} {vs_curr.upper()}{COL['nc']} ({symbol} {fmt_num(amount, 4)})"
|
|
)
|
|
print("-----------------------------------")
|
|
|
|
|
|
# ROUTING MATRIX
|
|
|
|
|
|
def show_help():
|
|
print("rate - multi-asset evaluation utility")
|
|
print("usage:")
|
|
print(" rate forex [amt] <base> [target] fiat configurations (alias: fx)")
|
|
print(" rate metal <gold|silver|ticker> precious spots contracts (alias: met)")
|
|
print(" rate energy <oil|brent|gas|ticker> fuel commodity options (alias: nrg)")
|
|
print(" rate indices <sp500|nasdaq|ticker> global market indices (alias: ind)")
|
|
print(" rate crypto <coin> [amt] [curr] digital tokens valuations (alias: c)")
|
|
print(" rate stock <ticker> [amt] [curr] corporate equities (alias: s)")
|
|
print(" rate etf <ticker> [amt] [curr] exchange traded funds (alias: e)")
|
|
sys.exit(0)
|
|
|
|
|
|
def main():
|
|
args = sys.argv[1:]
|
|
if not args or args[0] in ("-h", "--help", "help"):
|
|
show_help()
|
|
|
|
cmd = args[0].lower()
|
|
sub_args = args[1:]
|
|
|
|
if cmd in ("forex", "fx", "to", "from"):
|
|
if cmd == "to":
|
|
cmd_forex(["1", "NGN", sub_args[0]] if sub_args else ["NGN"])
|
|
elif cmd == "from":
|
|
cmd_forex(["1", sub_args[0], "NGN"] if sub_args else ["USD", "NGN"])
|
|
else:
|
|
cmd_forex(sub_args)
|
|
elif cmd in ("metal", "met"):
|
|
cmd_metal(sub_args)
|
|
elif cmd in ("energy", "nrg", "eng"):
|
|
cmd_energy(sub_args)
|
|
elif cmd in ("indices", "ind", "idx"):
|
|
cmd_indices(sub_args)
|
|
elif cmd in ("crypto", "c"):
|
|
cmd_crypto(sub_args)
|
|
elif cmd in ("stock", "stocks", "s"):
|
|
cmd_stock(sub_args)
|
|
elif cmd in ("etf", "e"):
|
|
cmd_etf(sub_args)
|
|
else:
|
|
# direct default fallback to legacy raw strings evaluation
|
|
cmd_forex(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|