256 lines
8.5 KiB
Python
256 lines
8.5 KiB
Python
import urllib.request
|
|
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
|
|
}
|
|
_history_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)",
|
|
}
|
|
|
|
MONTHS_MAP = {
|
|
"january": "01", "february": "02", "march": "03", "april": "04",
|
|
"may": "05", "june": "06", "july": "07", "august": "08",
|
|
"september": "09", "october": "10", "november": "11", "december": "12"
|
|
}
|
|
|
|
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 parse_date_iso(raw: str) -> str:
|
|
# Example: "25. August 2026"
|
|
try:
|
|
parts = raw.replace('.', '').split()
|
|
if len(parts) >= 3:
|
|
day = parts[0].zfill(2)
|
|
month = MONTHS_MAP.get(parts[1].lower(), "01")
|
|
year = parts[2]
|
|
return f"{year}-{month}-{day}"
|
|
except:
|
|
pass
|
|
return raw
|
|
|
|
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)]
|
|
if len(cells) >= 3 and cells[0]:
|
|
prices.append({
|
|
"metal": cells[0],
|
|
"name_fa": FA_NAMES.get(cells[0], cells[0]),
|
|
"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)]
|
|
if len(cells) >= 3 and cells[0]:
|
|
stocks.append({
|
|
"metal": cells[0],
|
|
"name_fa": FA_NAMES.get(cells[0], cells[0]),
|
|
"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"],
|
|
"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 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)]
|
|
if len(cells) >= 2 and cells[0]:
|
|
precious.append({
|
|
"name": cells[0],
|
|
"price_str": cells[1]
|
|
})
|
|
|
|
return {
|
|
"date": date_str or datetime.utcnow().strftime("%d. %B %Y"),
|
|
"updated_at": datetime.utcnow().isoformat(),
|
|
"metals": combined_metals,
|
|
"stocks": stocks,
|
|
"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_metal_history(metal_key: str, ttl_seconds: int = 300):
|
|
norm_key = metal_key.lower().strip()
|
|
field = FIELD_MAP.get(norm_key)
|
|
if not field:
|
|
# Fallback: check if the key already is a field
|
|
if norm_key.startswith("lme_"):
|
|
field = metal_key
|
|
else:
|
|
field = f"LME_{norm_key[:2].capitalize()}_cash"
|
|
|
|
now = time.time()
|
|
with _lock:
|
|
cached = _history_cache.get(field)
|
|
if cached and (now - cached["time"]) < ttl_seconds:
|
|
return cached["data"]
|
|
|
|
url = f'https://www.westmetall.com/en/markdaten.php?action=table&field={field}'
|
|
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 not tables:
|
|
return {"metal": metal_key, "field": field, "points": []}
|
|
|
|
rows = re.findall(r'<tr[^>]*>([\s\S]*?)</tr>', tables[0])
|
|
points = []
|
|
# Skip header
|
|
for r in rows[1:]:
|
|
cells = [re.sub(r'<.*?>', '', c).replace(' ', ' ').strip() for c in re.findall(r'<t[dh][^>]*>([\s\S]*?)</t[dh]>', r)]
|
|
if len(cells) >= 4 and cells[0]:
|
|
c_val = clean_num(cells[1])
|
|
tm_val = clean_num(cells[2])
|
|
s_val = clean_num(cells[3])
|
|
date_iso = parse_date_iso(cells[0])
|
|
points.append({
|
|
"date": date_iso,
|
|
"date_display": cells[0],
|
|
"cash": c_val,
|
|
"cash_str": cells[1],
|
|
"three_months": tm_val,
|
|
"three_months_str": cells[2],
|
|
"spread": round(tm_val - c_val, 2),
|
|
"stocks": s_val,
|
|
"stocks_str": cells[3],
|
|
})
|
|
|
|
# Reverse points to chronological order (oldest to newest) for charting
|
|
points.reverse()
|
|
|
|
result = {
|
|
"metal": metal_key,
|
|
"name_fa": FA_NAMES.get(metal_key.capitalize(), metal_key),
|
|
"field": field,
|
|
"total_points": len(points),
|
|
"points": points
|
|
}
|
|
|
|
with _lock:
|
|
_history_cache[field] = {"time": now, "data": result}
|
|
|
|
return result
|