-
Notifications
You must be signed in to change notification settings - Fork 2
/
Bhavcopy_BSE_Eq_01_01_2023_to_31_12_2023.py
361 lines (304 loc) · 15.7 KB
/
Bhavcopy_BSE_Eq_01_01_2023_to_31_12_2023.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
# -*- coding: utf-8 -*-
"""
Created on Sun Aug 18 21:37:55 2024
@author: ppare
"""
import os
import sys
import socket
import zipfile
import requests
import pandas as pd
from datetime import datetime, timedelta
from requests.packages.urllib3.exceptions import InsecureRequestWarning
# Suppress only the single InsecureRequestWarning from urllib3
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
def Download_BSE_Bhavcopy_File(BSE_Bhavcopy_URL, Bhavcopy_Download_Folder):
# Download a file from the given URL to the specified output folder
filename = os.path.basename(BSE_Bhavcopy_URL)
output_path = os.path.join(Bhavcopy_Download_Folder, filename)
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
try:
with requests.get(BSE_Bhavcopy_URL, stream=True, headers=headers, verify=False) as response:
response.raise_for_status() # Check if the request was successful
with open(output_path, 'wb') as file:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
file.write(chunk)
print(f'Eq-bhavcopy {filename} downloaded.')
return output_path
except requests.exceptions.RequestException as e:
print(formatted_date, "Bhavcopy file not available to download on BSE Server" )
return None
def Extract_BSE_Bhavcopy_Zip_Files(zip_file, Bhavcopy_Download_Folder):
# Extract files from the given ZIP file to the specified output folder
with zipfile.ZipFile(zip_file, 'r') as zip_ref:
zip_ref.extractall(Bhavcopy_Download_Folder)
print("Zip file extracted:", zip_file)
def Rename_BSE_Bhavcopy_Files(directory):
# Rename files in the specified directory to the desired format
files = os.listdir(directory)
renamed_files = []
for file in files:
if file.endswith(".csv"):
# Extract the date from the file name
date = file[-12:-4]
new_name = date + ".csv"
new_name = new_name.replace("BSE_EQ_BHAVCOPY_", "")
date_obj = datetime.strptime(date, "%d%m%Y")
new_name = date_obj.strftime("%Y-%m-%d") + "-BSE-EQ.csv"
# Rename the file
old_path = os.path.join(directory, file)
new_path = os.path.join(directory, new_name)
os.rename(old_path, new_path)
renamed_files.append((file, new_name))
else:
print(f"File '{file}' does not end with 'bhav.csv' and was not renamed.")
if renamed_files:
for old_name, new_name in renamed_files:
print(f"File '{old_name}' renamed to: '{new_name}'")
else:
print("No files found or no files meet the desired condition.")
def Modify_BSE_Bhavcopy_Files(directory):
# Modify the renamed files in the specified directory
files = os.listdir(directory)
for file in files:
if file.endswith("-BSE-EQ.csv"):
file_path = os.path.join(directory, file)
try:
# Read the CSV file
df = pd.read_csv(file_path)
# Remove the specified columns
columns_to_remove = [
"ISIN",
#"TckrSymb",
"FinInstrmId",
"FinInstrmNm",
#"SctySrs",
#"OpnPric",
#"HghPric",
#"LwPric",
#"ClsPric",
"LastPric",
"PrvsClsgPric",
#"TtlTradgVol",
"TtlTrfVal",
#"TradDt",
"TtlNbOfTxsExctd",
"FinInstrmTp",
"OffclCorpActnEvtId",
"RptgDt",
"TradRegnOrgn",
"MktTpandId",
"InstrmId",
"InstrmNm",
"FftyTwWkHgh",
"FftyTwWkLw",
"UnitOfMeasr",
"SttlmPric",
"AvrgPric",
"Ccy",
"Rsvd01",
"Rsvd02",
"Rsvd03",
"Rsvd04"
]
df = df.drop(columns=columns_to_remove)
Mutual_Fund_symbol_to_remove = [
'SENSEX1', 'SBISENSEX', 'CPSEETF', 'SENSEXBEES', 'SETFBSE100', 'UTISENSETF',
'HDFCSENSEX', 'BSLSENETFG', 'IDFSENSEXE', 'ICICIB22', 'BSE500IETF', 'SETFSN50',
'UTISXN50', 'SNXT50BEES', 'SENSEXADD', 'SENSEXETF', 'SENSEXIETF', 'MOM100',
'NIFTYIETF', 'NIF100IETF', 'NIF100BEES', 'NIFTY1', 'UTINIFTETF', 'LICNETFN50',
'HDFCNIFTY', 'LICNFNHGP', 'NV20IETF', 'MIDSELIETF', 'LOWVOLIETF', 'UTINEXT50',
'NEXT50IETF', 'NIFTYETF', 'ABSLNN50ET', 'BANKIETF', 'PVTBANIETF', 'NIESSPJ',
'NIESSPC', 'NIEHSPI', 'NIESSPE', 'NIEHSPD', 'NIEHSPE', 'NIESSPK', 'NIEHSPG',
'NIEHSPH', 'NIEHSPL', 'NIESSPL', 'NIESSPM', 'ABSLBANETF', 'MIDCAPIETF', 'NEXT50',
'08MPD', '08GPG', '11DPR', '11GPG', '11MPD', '11MPR', '11QPD', '11AGG', '11AMD',
'11DPD', 'ALPL30IETF', 'ITIETF', 'HDFCNIFBAN', 'UTIBANKETF', 'ESG', 'INFRABEES',
'MAFANG', 'HEALTHIETF', 'BFSI', 'FMCGIETF', 'AXISTECETF', 'AXISHCETF', 'AXISCETF',
'MASPTOP50', 'CONSUMIETF', 'EQUAL50ADD', 'MAHKTECH', 'MONQ50', 'MIDQ50ADD',
'NIFTY50ADD', 'AUTOIETF', 'MAKEINDIA', 'MOMOMENTUM', 'TECH', 'HEALTHY', 'BSLNIFTY',
'MIDCAPETF', 'MOLOWVOL', 'MOHEALTH', 'MOM30IETF', 'HDFCNIF100', 'HDFCNEXT50',
'INFRAIETF', 'NIFTYQLITY', 'MOMENTUM', 'MOVALUE', 'MOQUALITY', 'HDFCQUAL',
'HDFCGROWTH', 'HDFCVALUE', 'HDFCLOWVOL', 'HDFCMOMENT', 'HDFCNIFIT', 'HDFCPVTBAN',
'FINIETF', 'COMMOIETF', 'BANKETFADD', 'HDFCBSE500', 'HDFCSML250', 'HDFCMID150',
'PSUBNKIETF', 'AXSENSEX', 'LOWVOL', 'ITETFADD', 'BANKETF', 'PSUBANKADD',
'PVTBANKADD', 'QUAL30IETF', 'NIFMID150', 'NAVINIFTY', 'ITETF', 'ALPHAETF',
'NIFTYBETF', 'BANKBETF', 'NIFITETF', 'HDFCPSUBK', 'LICNMID100', 'SMALLCAP',
'MIDSMALL', 'MID150CASE', 'TOP100CASE', 'BBNPNBETF', 'EVINDIA', 'SBINEQWETF',
'OILIETF', 'ABSLPSE', 'NIFTYBEES', 'JUNIORBEES', 'BANKBEES', 'PSUBANK',
'PSUBNKBEES', 'SHARIABEES', 'QNIFTY', 'MOM50', 'BANKNIFTY1', 'SETFNIFBK',
'SETFNIF50'
]
# Filter out rows where 'TckrSymb' is in the list of values to remove
df = df[~df['TckrSymb'].isin(Mutual_Fund_symbol_to_remove)]
# Filter rows based on SERIES column
df = df[df['SctySrs'].isin(['A', 'B', 'M',"MS","MT","P","R","T","W","X","XT","Z","ZP"])]
# Convert the 'TradDt' column to datetime by inferring the format
df['TradDt'] = pd.to_datetime(df['TradDt'], errors='coerce')
# Check if there are any NaT (Not a Time) values which indicate conversion issues
if df['TradDt'].isna().any():
print("There are some dates that couldn't be converted:")
print(df[df['TradDt'].isna()])
# Convert the datetime object to the desired string format
df['TradDt'] = df['TradDt'].dt.strftime('%Y%m%d')
# Reorder the columns
cols = df.columns.tolist()
cols.remove('TradDt')
cols.insert(cols.index('OpnPric'), 'TradDt')
df = df[cols]
# Remove the SctySrs column
df = df.drop(columns=['SctySrs'])
# Sort the DataFrame by the 'TckrSymb' column
df = df.sort_values(by='TckrSymb')
# Save the modified dataframe back to CSV
df.to_csv(file_path, index=False, header=False)
print(f"{file} :Eq_Bhavcopy Data Structure converted to getbhavcopy")
except FileNotFoundError:
print(f"File not found: {file}")
def change_file_extension(Bhavcopy_Download_Folder, old_extension, new_extension):
# Iterate through all files in the folder
for filename in os.listdir(Bhavcopy_Download_Folder):
# Check if the file ends with the old extension
if filename.endswith("-BSE-EQ.csv"):
# Split the filename into name and extension, and replace the extension with the new extension
new_filename = os.path.splitext(filename)[0] + ".txt"
# Construct the full path of the old file
old_filepath = os.path.join(Bhavcopy_Download_Folder, filename)
# Construct the full path of the new file
new_filepath = os.path.join(Bhavcopy_Download_Folder, new_filename)
# Rename the old file to the new file
os.rename(old_filepath, new_filepath)
# Print a message to confirm the file extension change
#print(f"Changed Bhavcopy extension: {filename} to {new_filename}")
def copy_and_remove_files(Bhavcopy_Download_Folder, Final_Bhavcopy_Folder, remove_folders):
# Copy files to destination folder
for filename in os.listdir(Bhavcopy_Download_Folder):
if filename.endswith("-BSE-EQ.txt"):
source_filepath = os.path.join(Bhavcopy_Download_Folder, filename)
destination_filepath = os.path.join(Final_Bhavcopy_Folder, filename)
with open(source_filepath, 'rb') as source_file:
with open(destination_filepath, 'wb') as destination_file:
destination_file.write(source_file.read())
print(f"Bhavcopy file: {filename} copied to {Final_Bhavcopy_Folder}")
# Remove source folders
for folder in remove_folders:
for root, dirs, files in os.walk(folder, topdown=False):
for file in files:
os.remove(os.path.join(root, file))
for dir in dirs:
os.rmdir(os.path.join(root, dir))
os.rmdir(folder)
# Download files
Bhavcopy_Download_Folder = "C:/data_BSE"
if os.path.exists(Bhavcopy_Download_Folder):
# Remove the folder and its contents
for root, dirs, files in os.walk(Bhavcopy_Download_Folder, topdown=False):
for file in files:
os.remove(os.path.join(root, file))
for dir in dirs:
os.rmdir(os.path.join(root, dir))
os.rmdir(Bhavcopy_Download_Folder)
# Create the folder again
os.makedirs(Bhavcopy_Download_Folder)
# Final Bhavcopy folder
Final_Bhavcopy_Folder = "C:/Getbhavcopy_BSE"
if not os.path.exists(Final_Bhavcopy_Folder):
os.makedirs(Final_Bhavcopy_Folder)
#print(f"Created folder: {Final_Bhavcopy_Folder}")
"""
# Check if the folder exists and contains files
folder_path = "C:\\Getbhavcopy_BSE"
if os.path.isdir(folder_path) and os.listdir(folder_path):
# Get the list of files in the folder
file_list = [f for f in os.listdir(folder_path) if os.path.isfile(os.path.join(folder_path, f))]
# Extract the last date from the file names
last_date_str = max(f[:10] for f in file_list)
last_date = datetime.strptime(last_date_str, '%Y-%m-%d')
today = datetime.now().date()
if last_date.date() == today:
print()
print()
print("The Getbhavcopy_BSE database is up to date today",last_date.date(),"and no need to download any files.")
print()
print()
print("If you find my work valuable, please consider donating.")
print()
print()
print("💲 Donate via UPI: p.paresh25@oksbi")
print()
print()
sys.exit()
next_date = last_date + timedelta(days=1)
# Set the start date to the next day of the last date
start_date_str = next_date.strftime('%Y-%m-%d')
# Set the end date to the current date
end_date_str = datetime.today().strftime('%Y-%m-%d')
# Print the start and end dates
print(f"Start date: {start_date_str}")
print(f"End date: {end_date_str}")
print()
else:
# If the folder does not exist or is empty, get user input for the dates
start_date_str = input('Enter the start date (YYYY-MM-DD): ')
end_date_str = input('Enter the end date (YYYY-MM-DD): ')
"""
start_date_str = input('Enter the start date (YYYY-MM-DD): ')
end_date_str = input('Enter the end date (YYYY-MM-DD): ')
# Convert input dates to datetime objects
start_date = datetime.strptime(start_date_str, '%Y-%m-%d')
end_date = datetime.strptime(end_date_str, '%Y-%m-%d')
delta = end_date - start_date
date_range = []
for i in range(delta.days + 1):
date = start_date + timedelta(days=i)
if date.weekday() < 5: # Only include weekdays (Monday-Friday)
date_range.append(date)
socket.setdefaulttimeout(3)
downloaded_files = []
for date in date_range:
# Construct the URL for downloading the file
#https://www.bseindia.com/download/BhavCopy/Equity/BSE_EQ_BHAVCOPY_29122023.ZIP
BSE_Bhavcopy_URL = "https://www.bseindia.com/download/BhavCopy/Equity"
date_str = date.strftime("%d%m%Y")
filename = f"BSE_EQ_BHAVCOPY_{date_str}.ZIP"
BSE_Bhavcopy_URL = f"{BSE_Bhavcopy_URL}/{filename}"
formatted_date = date.strftime('%Y-%m-%d')
#print("URL:", BSE_Bhavcopy_URL)
try:
# Download the file and add it to the list of downloaded files
Downloaded_Bhavcopy_file = Download_BSE_Bhavcopy_File(BSE_Bhavcopy_URL, Bhavcopy_Download_Folder)
if Downloaded_Bhavcopy_file:
downloaded_files.append(Downloaded_Bhavcopy_file)
except Exception as e:
print(formatted_date, "Bhavcopy file not available to download on BSE Server")
print(e)
for file in downloaded_files:
try:
# Extract the downloaded ZIP files
Extract_BSE_Bhavcopy_Zip_Files(file, Bhavcopy_Download_Folder)
except Exception as e:
print("Error extracting file:", file)
print(e)
for file in downloaded_files:
# Remove the downloaded ZIP files
os.remove(file)
print("Removed:", file)
Rename_BSE_Bhavcopy_Files(Bhavcopy_Download_Folder)
Modify_BSE_Bhavcopy_Files(Bhavcopy_Download_Folder)
change_file_extension(Bhavcopy_Download_Folder, ".csv", ".txt")
remove_folders = ["C:/data_BSE"]
copy_and_remove_files(Bhavcopy_Download_Folder, Final_Bhavcopy_Folder, remove_folders)
print()
print()
print("All Eq-bhavcopy are saved to 'C:/Getbhavcopy_BSE'.")
print()
print()
print("If you find my work valuable, please consider donating.")
print()
print()
print("💲 Donate via UPI: p.paresh25@oksbi")
print()
print()