eshan6704 commited on
Commit
166165e
·
verified ·
1 Parent(s): a55bcda

Update CSV py

Browse files
Files changed (1) hide show
  1. CSV py +63 -108
CSV py CHANGED
@@ -1,134 +1,91 @@
1
  # CSV.py
 
2
  import pandas as pd
3
- from typing import List, Union, Optional
4
- from io import BytesIO
5
- import zipfile
6
  import requests
 
 
7
  from datetime import datetime as dt
 
 
8
  from persist import exists, load, save
9
 
10
 
11
- # ==========================
12
- # CSV / ZIP Loader
13
- # ==========================
14
  def load_csv(
15
- url_or_path: str,
16
  header_row: int = 0,
17
- drop_unnamed: bool = True,
18
- text_cols: Optional[List[str]] = None
19
  ) -> Union[pd.DataFrame, List[pd.DataFrame]]:
20
  """
21
- Load CSV or ZIP of CSVs.
22
- - If URL/path ends with .csv returns single DataFrame
23
- - If URL/path ends with .zip returns list of DataFrames (one per CSV inside)
24
- - header_row: row index for CSV header
25
- - drop_unnamed: drop unnamed columns
26
- - text_cols: columns to treat as text (optional)
27
  """
28
  text_cols = text_cols or []
29
 
30
- # --- ZIP case ---
31
- if url_or_path.lower().endswith(".zip"):
32
- if url_or_path.startswith("http"):
33
- r = requests.get(url_or_path)
34
- r.raise_for_status()
35
- z = zipfile.ZipFile(BytesIO(r.content))
36
- else:
37
- z = zipfile.ZipFile(url_or_path)
38
-
 
 
 
 
 
 
 
 
39
  dfs = []
40
  for name in z.namelist():
41
  if name.lower().endswith(".csv"):
42
  with z.open(name) as f:
43
  df = pd.read_csv(f, header=header_row)
44
- df = _clean_df(df, drop_unnamed, text_cols)
45
- dfs.append(df)
46
  return dfs
47
 
48
- # --- Single CSV case ---
49
- else:
50
- if url_or_path.startswith("http"):
51
- df = pd.read_csv(url_or_path, header=header_row)
52
- else:
53
- df = pd.read_csv(url_or_path, header=header_row)
54
- return _clean_df(df, drop_unnamed, text_cols)
55
-
56
-
57
- def _clean_df(df: pd.DataFrame, drop_unnamed: bool, text_cols: List[str]) -> pd.DataFrame:
58
- """Helper to clean columns, convert numeric, drop empty rows"""
59
- if drop_unnamed:
60
- df = df.loc[:, ~df.columns.str.contains("^Unnamed")]
61
-
62
- df.columns = (
63
- df.columns
64
- .str.strip()
65
- .str.replace(" ", "_")
66
- .str.replace("-", "_")
67
- )
68
 
69
- for col in df.columns:
70
- if col not in text_cols:
71
- df[col] = pd.to_numeric(df[col], errors="coerce")
72
-
73
- return df.dropna(how="all")
74
-
75
-
76
- # ==========================
77
- # HTML Generator
78
- # ==========================
79
- def df_to_html(
80
- df: pd.DataFrame,
81
- metric_col: Optional[str] = None,
82
- cache_key: Optional[str] = None
83
- ) -> str:
84
- """
85
- Convert DataFrame to HTML table with colored numeric values.
86
- - metric_col: highlights top 3 / bottom 3 for this column
87
- - cache_key: optional persist key for caching HTML
88
- """
89
- if df is None or df.empty:
90
- return "<p>No data available.</p>"
91
-
92
- # --- CACHE CHECK ---
93
- if cache_key and exists(cache_key, "html"):
94
- return load(cache_key, "html")
95
 
 
 
96
  df_html = df.copy()
97
- top3_up, top3_down = [], []
98
 
99
- if metric_col and metric_col in df_html.columns:
100
- col_numeric = pd.to_numeric(df_html[metric_col], errors="coerce").dropna()
101
- top3_up = col_numeric.nlargest(3).index.tolist()
102
- top3_down = col_numeric.nsmallest(3).index.tolist()
103
 
104
  for idx, row in df_html.iterrows():
105
  for col in df_html.columns:
106
  val = row[col]
107
  style = ""
108
  if isinstance(val, (int, float)):
109
- val_fmt = f"{val:.2f}"
110
  if val > 0:
111
  style = "pos"
112
  elif val < 0:
113
  style = "neg"
114
- if col == metric_col:
115
- if idx in top3_up:
116
  style += " top-up"
117
- elif idx in top3_down:
118
  style += " top-down"
119
- df_html.at[idx, col] = f'<span class="{style.strip()}">{val_fmt}</span>'
120
  else:
121
  df_html.at[idx, col] = str(val)
122
 
123
- html_out = f"""
124
  <!DOCTYPE html>
125
  <html>
126
  <head>
127
  <meta charset="UTF-8">
128
- <title>CSV Table</title>
129
  <style>
130
  body {{ font-family: Arial; background:#f5f5f5; padding:12px; }}
131
- table {{ border-collapse: collapse; width: 100%; background:white; }}
132
  th, td {{ border:1px solid #bbb; padding:6px; font-size:13px; }}
133
  th {{ background:#222; color:white; }}
134
  .pos {{ color:green; font-weight:bold; }}
@@ -138,42 +95,40 @@ th {{ background:#222; color:white; }}
138
  </style>
139
  </head>
140
  <body>
 
141
  {df_html.to_html(index=False, escape=False)}
142
  </body>
143
  </html>
144
  """
145
 
146
- # --- SAVE CACHE ---
147
- if cache_key:
148
- save(cache_key, html_out, "html")
149
-
150
- return html_out
151
-
152
 
153
- # ==========================
154
- # NSE High-Low Master
155
- # ==========================
156
- def nse_highlow(date_str: str = None) -> str:
157
  """
158
- Master function to fetch NSE High-Low CSV and return HTML.
159
- - date_str: "DD-MM-YYYY", default today
160
- - uses caching (persist.py) for HTML
 
161
  """
162
- if date_str is None:
163
  date_str = dt.now().strftime("%d-%m-%Y")
164
 
165
- # Cache key
166
  cache_key = f"highlow_{date_str}"
167
 
168
- # --- URL for NSE High-Low CSV ---
169
- dt_obj = dt.strptime(date_str, "%d-%m-%Y")
170
- url_date = dt_obj.strftime("%d%m%Y")
171
- url = f"https://archives.nseindia.com/content/indices/ind_close_all_{url_date}.csv"
172
 
173
- # --- Load CSV using load_csv ---
174
- df = load_csv(url, header_row=2, text_cols=["Index_Name", "Index_Date"])
 
 
 
175
 
176
- # --- Generate HTML using df_to_html ---
177
- html = df_to_html(df, metric_col="PERCENT_CHANGE", cache_key=cache_key)
 
 
 
178
 
 
 
179
  return html
 
1
  # CSV.py
2
+
3
  import pandas as pd
 
 
 
4
  import requests
5
+ import zipfile
6
+ from io import BytesIO
7
  from datetime import datetime as dt
8
+ from typing import List, Union
9
+
10
  from persist import exists, load, save
11
 
12
 
 
 
 
13
  def load_csv(
14
+ url: str,
15
  header_row: int = 0,
16
+ text_cols: List[str] | None = None
 
17
  ) -> Union[pd.DataFrame, List[pd.DataFrame]]:
18
  """
19
+ Load CSV or ZIP containing CSVs
20
+ - .csv -> DataFrame
21
+ - .zip -> List[DataFrame]
 
 
 
22
  """
23
  text_cols = text_cols or []
24
 
25
+ def _clean_df(df: pd.DataFrame) -> pd.DataFrame:
26
+ df = df.loc[:, ~df.columns.str.contains("^Unnamed")]
27
+ df.columns = (
28
+ df.columns
29
+ .str.strip()
30
+ .str.replace(" ", "_")
31
+ .str.replace("-", "_")
32
+ )
33
+ for col in df.columns:
34
+ if col not in text_cols:
35
+ df[col] = pd.to_numeric(df[col], errors="coerce")
36
+ return df.dropna(how="all")
37
+
38
+ if url.lower().endswith(".zip"):
39
+ r = requests.get(url)
40
+ r.raise_for_status()
41
+ z = zipfile.ZipFile(BytesIO(r.content))
42
  dfs = []
43
  for name in z.namelist():
44
  if name.lower().endswith(".csv"):
45
  with z.open(name) as f:
46
  df = pd.read_csv(f, header=header_row)
47
+ dfs.append(_clean_df(df))
 
48
  return dfs
49
 
50
+ df = pd.read_csv(url, header=header_row)
51
+ return _clean_df(df)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
52
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
53
 
54
+ def _highlow_html_formatter(df: pd.DataFrame, date_str: str) -> str:
55
+ metric = "PERCENT_CHANGE"
56
  df_html = df.copy()
 
57
 
58
+ top_up = df[metric].nlargest(3).index if metric in df else []
59
+ top_dn = df[metric].nsmallest(3).index if metric in df else []
 
 
60
 
61
  for idx, row in df_html.iterrows():
62
  for col in df_html.columns:
63
  val = row[col]
64
  style = ""
65
  if isinstance(val, (int, float)):
66
+ txt = f"{val:.2f}"
67
  if val > 0:
68
  style = "pos"
69
  elif val < 0:
70
  style = "neg"
71
+ if col == metric:
72
+ if idx in top_up:
73
  style += " top-up"
74
+ elif idx in top_dn:
75
  style += " top-down"
76
+ df_html.at[idx, col] = f'<span class="{style.strip()}">{txt}</span>'
77
  else:
78
  df_html.at[idx, col] = str(val)
79
 
80
+ return f"""
81
  <!DOCTYPE html>
82
  <html>
83
  <head>
84
  <meta charset="UTF-8">
85
+ <title>NSE High-Low {date_str}</title>
86
  <style>
87
  body {{ font-family: Arial; background:#f5f5f5; padding:12px; }}
88
+ table {{ border-collapse: collapse; width:100%; background:white; }}
89
  th, td {{ border:1px solid #bbb; padding:6px; font-size:13px; }}
90
  th {{ background:#222; color:white; }}
91
  .pos {{ color:green; font-weight:bold; }}
 
95
  </style>
96
  </head>
97
  <body>
98
+ <h2>NSE Index High / Low — {date_str}</h2>
99
  {df_html.to_html(index=False, escape=False)}
100
  </body>
101
  </html>
102
  """
103
 
 
 
 
 
 
 
104
 
105
+ def nse_highlow(date_str: str | None = None) -> str:
 
 
 
106
  """
107
+ Master NSE High-Low function
108
+ - Uses load_csv
109
+ - Builds HTML
110
+ - Persists ONLY HTML
111
  """
112
+ if not date_str:
113
  date_str = dt.now().strftime("%d-%m-%Y")
114
 
 
115
  cache_key = f"highlow_{date_str}"
116
 
117
+ if exists(cache_key, "html"):
118
+ return load(cache_key, "html")
 
 
119
 
120
+ d = dt.strptime(date_str, "%d-%m-%Y")
121
+ url = (
122
+ "https://archives.nseindia.com/content/indices/"
123
+ f"ind_close_all_{d.strftime('%d%m%Y')}.csv"
124
+ )
125
 
126
+ df = load_csv(
127
+ url=url,
128
+ header_row=2,
129
+ text_cols=["Index_Name", "Index_Date"]
130
+ )
131
 
132
+ html = _highlow_html_formatter(df, date_str)
133
+ save(cache_key, html, "html")
134
  return html