296 lines
10 KiB
Python
296 lines
10 KiB
Python
import urllib.request
|
||
import xml.etree.ElementTree as ET
|
||
import re
|
||
from datetime import datetime
|
||
import threading
|
||
import time
|
||
|
||
HEADERS = {
|
||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
|
||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
|
||
'Accept-Language': 'en-US,en;q=0.5'
|
||
}
|
||
|
||
_cache = {
|
||
"data": None,
|
||
"last_fetched": 0
|
||
}
|
||
_diagram_cache = {}
|
||
_lock = threading.Lock()
|
||
|
||
FIELD_MAP = {
|
||
"copper": "LME_Cu_cash",
|
||
"cu": "LME_Cu_cash",
|
||
"aluminium": "LME_Al_cash",
|
||
"al": "LME_Al_cash",
|
||
"aluminum": "LME_Al_cash",
|
||
"zinc": "LME_Zn_cash",
|
||
"zn": "LME_Zn_cash",
|
||
"nickel": "LME_Ni_cash",
|
||
"ni": "LME_Ni_cash",
|
||
"lead": "LME_Pb_cash",
|
||
"pb": "LME_Pb_cash",
|
||
"tin": "LME_Sn_cash",
|
||
"sn": "LME_Sn_cash",
|
||
}
|
||
|
||
FA_NAMES = {
|
||
"Copper": "مس (Copper)",
|
||
"Tin": "قلع (Tin)",
|
||
"Lead": "سرب (Lead)",
|
||
"Zinc": "روی (Zinc)",
|
||
"Aluminium": "آلومینیوم (Aluminium)",
|
||
"Nickel": "نیکل (Nickel)",
|
||
"WM_Cu_low": "مس کاتد آلمان (WM-Notiz کف)",
|
||
"WM_Cu_high": "مس کاتد آلمان (WM-Notiz سقف)",
|
||
"WI_Cu": "مس ویلند (Wieland Copper)",
|
||
"ACI": "شاخص مس پیشرفته (ACI)",
|
||
"MB_bronze_94_6": "برنز 94/6",
|
||
"MB_MS_58_1": "برنج MS 58 (مرحله ۱)",
|
||
"MB_MS_58_2": "برنج MS 58 (مرحله ۲)",
|
||
"MB_MS_63_37": "برنج MS 63/37",
|
||
"MB_MS_63_wire": "مفتول برنجی MS 63",
|
||
"USD_ozt_London": "انس طلای لندن (Fixing)",
|
||
"Au": "طلای شمش در اروپا",
|
||
"Au_processed": "طلای ساختهشده در اروپا",
|
||
"Ag": "نقره خالص در اروپا",
|
||
"Ag_processed": "نقره ساختهشده در اروپا",
|
||
}
|
||
|
||
def clean_num(val: str) -> float:
|
||
if not val:
|
||
return 0.0
|
||
val = val.replace(',', '').replace('+', '').replace('$', '').strip()
|
||
try:
|
||
return float(val)
|
||
except:
|
||
return 0.0
|
||
|
||
def fetch_raw():
|
||
url = 'https://www.westmetall.com/en/markdaten.php'
|
||
req = urllib.request.Request(url, headers=HEADERS)
|
||
with urllib.request.urlopen(req, timeout=12) as resp:
|
||
html = resp.read().decode('utf-8', errors='ignore')
|
||
|
||
tables = re.findall(r'<table[^>]*>([\s\S]*?)</table>', html)
|
||
if len(tables) < 2:
|
||
return None
|
||
|
||
# Table 0: Official LME Prices
|
||
t0_rows = re.findall(r'<tr[^>]*>([\s\S]*?)</tr>', tables[0])
|
||
date_str = ""
|
||
prices = []
|
||
if t0_rows:
|
||
header_cells = re.findall(r'<t[dh][^>]*>([\s\S]*?)</t[dh]>', t0_rows[0])
|
||
if len(header_cells) > 1:
|
||
date_str = re.sub(r'<.*?>', '', header_cells[1]).strip()
|
||
|
||
for r in t0_rows[2:]:
|
||
cells = [re.sub(r'<.*?>', '', c).replace(' ', ' ').strip() for c in re.findall(r'<t[dh][^>]*>([\s\S]*?)</t[dh]>', r)]
|
||
links = re.findall(r'field=([a-zA-Z0-9_]+)', r)
|
||
field = links[0] if links else f"LME_{cells[0][:2]}_cash"
|
||
if len(cells) >= 3 and cells[0]:
|
||
prices.append({
|
||
"metal": cells[0],
|
||
"name_fa": FA_NAMES.get(cells[0], cells[0]),
|
||
"field": field,
|
||
"cash": clean_num(cells[1]),
|
||
"cash_str": cells[1],
|
||
"three_months": clean_num(cells[2]),
|
||
"three_months_str": cells[2],
|
||
"unit": "USD/mt"
|
||
})
|
||
|
||
# Table 1: LME Stocks
|
||
t1_rows = re.findall(r'<tr[^>]*>([\s\S]*?)</tr>', tables[1])
|
||
stocks = []
|
||
if len(t1_rows) > 2:
|
||
for r in t1_rows[2:]:
|
||
cells = [re.sub(r'<.*?>', '', c).replace(' ', ' ').strip() for c in re.findall(r'<t[dh][^>]*>([\s\S]*?)</t[dh]>', r)]
|
||
links = re.findall(r'field=([a-zA-Z0-9_]+)', r)
|
||
field = links[0] if links else ""
|
||
if len(cells) >= 3 and cells[0]:
|
||
stocks.append({
|
||
"metal": cells[0],
|
||
"name_fa": FA_NAMES.get(cells[0], cells[0]),
|
||
"field": field,
|
||
"stocks": clean_num(cells[1]),
|
||
"stocks_str": cells[1],
|
||
"change": clean_num(cells[2]),
|
||
"change_str": cells[2],
|
||
"unit": "mt"
|
||
})
|
||
|
||
# Combine Prices + Stocks by metal
|
||
combined_metals = []
|
||
stocks_map = {s["metal"]: s for s in stocks}
|
||
for p in prices:
|
||
m_name = p["metal"]
|
||
s_data = stocks_map.get(m_name, {})
|
||
spread = round(p["three_months"] - p["cash"], 2)
|
||
combined_metals.append({
|
||
"metal": m_name,
|
||
"name_fa": p["name_fa"],
|
||
"field": p["field"],
|
||
"cash_settlement": p["cash"],
|
||
"cash_str": p["cash_str"],
|
||
"three_months": p["three_months"],
|
||
"three_months_str": p["three_months_str"],
|
||
"spread": spread,
|
||
"spread_str": f"{spread:+,.2f}",
|
||
"stocks": s_data.get("stocks", 0),
|
||
"stocks_str": s_data.get("stocks_str", "-"),
|
||
"stocks_change": s_data.get("change", 0),
|
||
"stocks_change_str": s_data.get("change_str", "-"),
|
||
"unit": "USD/mt"
|
||
})
|
||
|
||
# Table 2: FX Rates
|
||
fx = []
|
||
if len(tables) > 2:
|
||
for r in re.findall(r'<tr[^>]*>([\s\S]*?)</tr>', tables[2])[1:]:
|
||
cells = [re.sub(r'<.*?>', '', c).replace(' ', ' ').strip() for c in re.findall(r'<t[dh][^>]*>([\s\S]*?)</t[dh]>', r)]
|
||
if len(cells) >= 2 and cells[0]:
|
||
fx.append({
|
||
"pair": cells[0],
|
||
"rate": clean_num(cells[1]),
|
||
"rate_str": cells[1]
|
||
})
|
||
|
||
# Table 3: German Metal Quotations
|
||
german_symbols = []
|
||
if len(tables) > 3:
|
||
for r in re.findall(r'<tr[^>]*>([\s\S]*?)</tr>', tables[3])[2:]:
|
||
cells = [re.sub(r'<.*?>', '', c).replace(' ', ' ').strip() for c in re.findall(r'<t[dh][^>]*>([\s\S]*?)</t[dh]>', r)]
|
||
links = re.findall(r'field=([a-zA-Z0-9_]+)', r)
|
||
field = links[0] if links else ""
|
||
if len(cells) >= 3 and cells[0]:
|
||
german_symbols.append({
|
||
"name": cells[0],
|
||
"name_fa": FA_NAMES.get(field, cells[0]),
|
||
"field": field,
|
||
"price_str": cells[1],
|
||
"prev_str": cells[2],
|
||
"unit": "EUR/100kg"
|
||
})
|
||
|
||
# Table 4: Precious Metals
|
||
precious = []
|
||
if len(tables) > 4:
|
||
for r in re.findall(r'<tr[^>]*>([\s\S]*?)</tr>', tables[4])[1:]:
|
||
cells = [re.sub(r'<.*?>', '', c).replace(' ', ' ').strip() for c in re.findall(r'<t[dh][^>]*>([\s\S]*?)</t[dh]>', r)]
|
||
links = re.findall(r'field=([a-zA-Z0-9_]+)', r)
|
||
field = links[0] if links else ""
|
||
if len(cells) >= 2 and cells[0]:
|
||
precious.append({
|
||
"name": cells[0],
|
||
"name_fa": FA_NAMES.get(field, cells[0]),
|
||
"field": field,
|
||
"price_str": cells[1],
|
||
"prev_str": cells[2] if len(cells) > 2 else "-"
|
||
})
|
||
|
||
return {
|
||
"date": date_str or datetime.utcnow().strftime("%d. %B %Y"),
|
||
"updated_at": datetime.utcnow().isoformat(),
|
||
"metals": combined_metals,
|
||
"stocks": stocks,
|
||
"german_symbols": german_symbols,
|
||
"exchange_rates": fx,
|
||
"precious_metals": precious
|
||
}
|
||
|
||
def get_lme_data(ttl_seconds: int = 180):
|
||
now = time.time()
|
||
with _lock:
|
||
if _cache["data"] and (now - _cache["last_fetched"]) < ttl_seconds:
|
||
return _cache["data"]
|
||
try:
|
||
fresh = fetch_raw()
|
||
if fresh:
|
||
_cache["data"] = fresh
|
||
_cache["last_fetched"] = now
|
||
return fresh
|
||
except Exception as e:
|
||
if _cache["data"]:
|
||
return _cache["data"]
|
||
raise e
|
||
return _cache["data"]
|
||
|
||
def get_entire_diagram_data(field: str, ttl_seconds: int = 3600):
|
||
"""
|
||
Fetches the ENTIRE multi-year historical dataset for any symbol
|
||
from the official XML marketdata API.
|
||
"""
|
||
field_clean = field.strip()
|
||
# Resolve aliases (e.g. "copper" -> "LME_Cu_cash")
|
||
field_clean = FIELD_MAP.get(field_clean.lower(), field_clean)
|
||
|
||
now = time.time()
|
||
with _lock:
|
||
cached = _diagram_cache.get(field_clean)
|
||
if cached and (now - cached["time"]) < ttl_seconds:
|
||
return cached["data"]
|
||
|
||
url = f'https://www.westmetall.com/api/marketdata/en/{field_clean}/'
|
||
req = urllib.request.Request(url, headers=HEADERS)
|
||
with urllib.request.urlopen(req, timeout=15) as r:
|
||
xml_text = r.read().decode('utf-8')
|
||
|
||
root = ET.fromstring(xml_text)
|
||
series_elem = root.find('series')
|
||
if series_elem is None:
|
||
return {"field": field_clean, "lines": []}
|
||
|
||
lines = []
|
||
for line in series_elem:
|
||
raw_name = line.attrib.get('name', field_clean)
|
||
color = line.attrib.get('lineColor', '#D02E00')
|
||
unit = line.attrib.get('yUnit', '')
|
||
precision = int(line.attrib.get('yPrecision', '2'))
|
||
|
||
pts = []
|
||
for val in line:
|
||
raw_x = val.attrib.get('x', '') # YYYY/MM/DD
|
||
date_iso = raw_x.replace('/', '-')
|
||
# Also format DD/MM/YYYY for tooltips as shown in user screenshot
|
||
parts = date_iso.split('-')
|
||
date_display = f"{parts[2]}/{parts[1]}/{parts[0]}" if len(parts) == 3 else date_iso
|
||
|
||
y_str = val.attrib.get('y', '0').replace(',', '')
|
||
try:
|
||
y_val = float(y_str)
|
||
except:
|
||
y_val = 0.0
|
||
|
||
pts.append({
|
||
"x": date_iso,
|
||
"date_display": date_display,
|
||
"y": y_val
|
||
})
|
||
|
||
lines.append({
|
||
"name": raw_name,
|
||
"color": color,
|
||
"unit": unit,
|
||
"precision": precision,
|
||
"count": len(pts),
|
||
"points": pts
|
||
})
|
||
|
||
result = {
|
||
"field": field_clean,
|
||
"name_fa": FA_NAMES.get(field_clean, field_clean),
|
||
"total_points": lines[0]["count"] if lines else 0,
|
||
"lines": lines
|
||
}
|
||
|
||
with _lock:
|
||
_diagram_cache[field_clean] = {"time": now, "data": result}
|
||
|
||
return result
|
||
|
||
# Backward compatibility alias
|
||
def get_metal_history(metal_key: str, ttl_seconds: int = 300):
|
||
return get_entire_diagram_data(metal_key, ttl_seconds)
|