update list

list.py

@@ -1,26 +1,29 @@
 # -*- coding: utf-8 -*-
+"""
+Minimal API for PY-Trade filters & companies
+- /getfilters                      -> countries -> exchanges -> indices
+- /getcompanies?code=NIFTY50       -> { code, asOf, count, constituents[] }
+"""
 
 from __future__ import annotations
-import csv, io, json, time
-from typing import Dict, List, Any, Optional
+import csv, io, json, time, os
+from typing import Dict, List, Any
 from pathlib import Path
-from io import StringIO
 
 import requests
-
-# optional (for Wikipedia tables)
-try:
-    import pandas as pd  # requires: pip install pandas lxml
-    HAS_PANDAS = True
-except Exception:
-    HAS_PANDAS = False
+from flask import Flask, request, jsonify
+from flask_cors import CORS
 
 # ---------- configuration ----------
 UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127 Safari/537.36"
 REFERER = "https://www.niftyindices.com/indices/equity/broad-based-indices"
-TTL_SECONDS = 60 * 60 * 12  # 12 hours
-CACHE_DIR = Path(__file__).with_name("cache")
-CACHE_DIR.mkdir(exist_ok=True)
+TTL_SECONDS = 60 * 60 * 12  # 12h cache
+# Use /data/cache on HF Spaces (writable), else fall back to local folder "cache" next to this file.
+DEFAULT_CACHE_DIR = os.getenv("CACHE_DIR", "/data/cache")
+CACHE_DIR = Path(DEFAULT_CACHE_DIR if DEFAULT_CACHE_DIR else ".").expanduser()
+if CACHE_DIR == Path("."):
+    CACHE_DIR = Path(__file__).with_name("cache")
+CACHE_DIR.mkdir(parents=True, exist_ok=True)
 
 # Official CSV endpoints for NSE indices
 NIFTY_URLS: Dict[str, str] = {
@@ -31,15 +34,8 @@ NIFTY_URLS: Dict[str, str] = {
     "NIFTY500":     "https://www.niftyindices.com/IndexConstituent/ind_nifty500list.csv",
 }
 
-# Wikipedia pages for other markets
-WIKI_PAGES: Dict[str, str] = {
-    "NASDAQ100": "https://en.wikipedia.org/wiki/NASDAQ-100",
-    "DAX40":     "https://en.wikipedia.org/wiki/DAX",
-    "OMXS30":    "https://en.wikipedia.org/wiki/OMX_Stockholm_30",
-}
-
-# Filters payload
-_MARKETS: Dict[str, Dict[str, List[Dict[str, str]]]] = {
+# Filters payload for the UI (add more countries/exchanges here later)
+MARKETS: Dict[str, Dict[str, List[Dict[str, str]]]] = {
     "India": {
         "NSE (National Stock Exchange)": [
             {"code": "NIFTY50",     "name": "NIFTY 50"},
@@ -48,86 +44,34 @@ _MARKETS: Dict[str, Dict[str, List[Dict[str, str]]]] = {
             {"code": "NIFTYMID100", "name": "NIFTY Midcap 100"},
             {"code": "NIFTY500",    "name": "NIFTY 500"},
         ]
-    },
-    "United States": {
-        "NASDAQ": [
-            {"code": "NASDAQ100", "name": "NASDAQ-100"}
-        ]
-    },
-    "Germany": {
-        "XETRA (Deutsche Börse)": [
-            {"code": "DAX40", "name": "DAX 40"}
-        ]
-    },
-    "Sweden": {
-        "OMX Stockholm": [
-            {"code": "OMXS30", "name": "OMX Stockholm 30"}
-        ]
     }
 }
 
-# ---------- public API (for routes) ----------
-def get_markets() -> Dict[str, Dict[str, List[Dict[str, str]]]]:
-    """Return filters structure used by UI."""
-    return _MARKETS
-
-def build_companies_payload(code: str) -> Dict[str, Any]:
-    """Return standardized payload for an index (with caching)."""
-    code = (code or "").upper().strip()
-    if not code:
-        raise ValueError("Index code is required.")
-
-    # try cache
-    cached = _load_cache(code)
-    if cached:
-        return cached
-
-    # route per code
-    if code in NIFTY_URLS:
-        url = NIFTY_URLS[code]
-        text = _http_get_text(url)
-        rows = _parse_nifty_csv(text)
-        exchange, country, currency, source = "NSE", "IN", "INR", url
-
-    elif code == "NASDAQ100":
-        rows, exchange, country, currency, source = _parse_nasdaq100()
-
-    elif code == "DAX40":
-        rows, exchange, country, currency, source = _parse_dax40()
-
-    elif code == "OMXS30":
-        rows, exchange, country, currency, source = _parse_omxs30()
-
-    else:
-        raise ValueError(f"Unknown index code: {code}")
-
-    payload = {
-        "code": code,
-        "exchange": exchange,
-        "country": country,
-        "currency": currency,
-        "asOf": _now_iso_utc(),
-        "count": len(rows),
-        "constituents": rows,
-        "source": source,
-    }
-    _save_cache(code, payload)
-    return payload
-
-# ---------- internals ----------
-def _http_get_text(url: str, accept: str = "text/csv,*/*") -> str:
+# ---------- utilities ----------
+def http_get_text(url: str) -> str:
     sess = requests.Session()
-    sess.headers.update({"User-Agent": UA, "Referer": REFERER, "Accept": accept})
-    r = sess.get(url, timeout=30)
+    sess.headers.update({"User-Agent": UA, "Referer": REFERER, "Accept": "text/csv,*/*"})
+    r = sess.get(url, timeout=25)
     r.raise_for_status()
     r.encoding = r.encoding or "utf-8"
     return r.text
 
-def _cache_path(code: str) -> Path:
+def parse_nifty_csv(text: str) -> List[Dict[str, str]]:
+    # Columns: Company Name, Industry, Symbol, Series, ISIN Code
+    out: List[Dict[str, str]] = []
+    rdr = csv.DictReader(io.StringIO(text))
+    for row in rdr:
+        sym = (row.get("Symbol") or "").strip()
+        name = (row.get("Company Name") or "").strip()
+        if sym and name:
+            out.append({"symbol": f"{sym}.NS", "company": name})
+    return out
+
+def cache_path(code: str) -> Path:
     return CACHE_DIR / f"{code.lower()}.json"
 
-def _load_cache(code: str) -> Optional[Any]:
-    fp = _cache_path(code)
+def load_cache(code: str) -> Any | None:
+    fp = cache_path(code)
     if not fp.exists():
         return None
     age = time.time() - fp.stat().st_mtime
@@ -136,86 +80,35 @@ def _load_cache(code: str) -> Optional[Any]:
     with fp.open("r", encoding="utf-8") as f:
         return json.load(f)
 
-def _save_cache(code: str, payload: Any) -> None:
-    fp = _cache_path(code)
+def save_cache(code: str, payload: Any) -> None:
+    fp = cache_path(code)
     with fp.open("w", encoding="utf-8") as f:
         json.dump(payload, f, ensure_ascii=False, indent=2)
 
-def _now_iso_utc() -> str:
-    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
-
-def _parse_nifty_csv(text: str) -> List[Dict[str, str]]:
-    out: List[Dict[str, str]] = []
-    rdr = csv.DictReader(io.StringIO(text))
-    for row in rdr:
-        sym = (row.get("Symbol") or "").strip()
-        name = (row.get("Company Name") or "").strip()
-        if sym and name:
-            out.append({"symbol": f"{sym}.NS", "company": name})
-    return out
-
-# ---- Wikipedia helpers ----
-def _fetch_wiki_tables(url: str):
-    if not HAS_PANDAS:
-        raise RuntimeError("pandas/lxml not installed. Run: pip install pandas lxml")
-    html = _http_get_text(url, accept="text/html,*/*")
-    return pd.read_html(StringIO(html))
-
-def _pick_table_and_columns(tables, ticker_candidates, company_candidates):
-    for t in tables:
-        cols_map = {str(c).strip().lower(): c for c in t.columns}
-        ticker_col = next((cols_map[c] for c in ticker_candidates if c in cols_map), None)
-        company_col = next((cols_map[c] for c in company_candidates if c in cols_map), None)
-        if ticker_col is not None and company_col is not None:
-            return t, ticker_col, company_col
-    raise RuntimeError(
-        f"No suitable table found. Ticker in {ticker_candidates}, company in {company_candidates}."
-    )
-
-def _parse_wiki_constituents(url: str, ticker_candidates, company_candidates, suffix: str, upper_tickers: bool) -> List[Dict[str, str]]:
-    tables = _fetch_wiki_tables(url)
-    df, t_col, c_col = _pick_table_and_columns(tables, ticker_candidates, company_candidates)
-    rows: List[Dict[str, str]] = []
-    for sym, name in zip(df[t_col], df[c_col]):
-        s = str(sym).strip()
-        if not s or not str(name).strip():
-            continue
-        if upper_tickers:
-            s = s.upper()
-        rows.append({"symbol": f"{s}{suffix}", "company": str(name).strip()})
-    if not rows:
-        raise RuntimeError("Parsed zero rows from Wikipedia table.")
-    return rows
+def build_companies_payload(code: str) -> Dict[str, Any]:
+    code = code.upper()
+    # 1) try cache
+    cached = load_cache(code)
+    if cached:
+        return cached
 
-def _parse_nasdaq100():
-    url = WIKI_PAGES["NASDAQ100"]
-    rows = _parse_wiki_constituents(
-        url,
-        ticker_candidates=["ticker", "symbol"],
-        company_candidates=["company", "name"],
-        suffix="",
-        upper_tickers=True,
-    )
-    return rows, "NASDAQ", "US", "USD", url
+    # 2) fetch official CSV
+    url = NIFTY_URLS.get(code)
+    if not url:
+        raise ValueError(f"Unknown index code: {code}")
+    text = http_get_text(url)
+    rows = parse_nifty_csv(text)
 
-def _parse_dax40():
-    url = WIKI_PAGES["DAX40"]
-    rows = _parse_wiki_constituents(
-        url,
-        ticker_candidates=["ticker symbol", "ticker", "symbol"],
-        company_candidates=["company", "name"],
-        suffix=".DE",
-        upper_tickers=True,
-    )
-    return rows, "XETRA", "DE", "EUR", url
+    payload = {
+        "code": code,
+        "exchange": "NSE",
+        "country": "IN",
+        "currency": "INR",
+        "asOf": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "count": len(rows),
+        "constituents": rows,
+        "source": url,
+    }
+    save_cache(code, payload)
+    return payload
 
-def _parse_omxs30():
-    url = WIKI_PAGES["OMXS30"]
-    rows = _parse_wiki_constituents(
-        url,
-        ticker_candidates=["ticker", "symbol"],
-        company_candidates=["company", "name"],
-        suffix=".ST",
-        upper_tickers=True,
-    )
-    return rows, "OMX Stockholm", "SE", "SEK", url