<jupyter_start><jupyter_text>Scrape HSI Constituents from official websiteThis script is used to construct a time series Hang Seng Index universe based on customized date listThree steps are involved:1. Scrape raw files from Hang Seng website (HSI Official Website)2. Clean the raw files3. Construct a time series universe based on the cleaned raw files.Scrapy is preferred for web scraping in most cases but not in this case because it can't easily scrape with JavaScript. In this case, we need to click 2 JavaScript buttons to download below 2 raw files.1. Latest Constituents2. Historical Change in HSIWith above 2 files, we could easily construct the time series universe. Cleaned and uncleaned raw files will be showing as well. Another key note is HK Date List. Our date list must in HK trading calendar. Only HK trading date will be considered as valid input when downloading latest constituent file from the website.Finally, we will generate a csv file named hsiUniverseDf.- We must put geckodriver.exe into the bin folder under our python folder in order to use Firefox as a webdriver under selenium. (I'm using Mac)<jupyter_code>import pandas as pd
import numpy as np
from selenium import webdriver
from selenium.webdriver.support.ui import Select
from selenium.webdriver.common.by import By
from selenium.webdriver.firefox.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from dateutil.relativedelta import relativedelta
from datetime import date
import os
import re
import shutil
import time
import holidays
import glob
def getHKDateList(startDate, endDate, frequency):
    # Get dateList according to HK Calendar
    dateList = pd.date_range(start=startDate, end=endDate, freq=frequency)
    hkHolidays = holidays.HK()
    dateList = [date if date not in hkHolidays else 
                max([loopDate for loopDate in pd.date_range(start=date-relativedelta(days=14), end=date, freq='D') if 
                     loopDate not in hkHolidays and loopDate.weekday() != 5 and loopDate.weekday() != 6]) 
                for date in dateList]
    del hkHolidays
    return dateList
    
def downloadRawFiles(downloadPath):
    print(downloadPath)
    print('Start downloading HSI files to above path.')
    
    # Create dowload folder if it's not exist
    if os.path.isdir(downloadPath) == False:
        os.mkdir(downloadPath)
        print('Created ' + downloadPath + ' as it is not exist')
    
    # Clean download folder if past files exist
    oldFileList = [downloadPath + "/" + f for f in os.listdir(downloadPath) if f == 'hsiCons.csv' or f == 'hsiConsChange.xlsx']
    
    if oldFileList != []:
        for file in oldFileList:
            os.remove(file)
    
    # Get Latest Month HK Date List
    latestDate = date.today() - relativedelta(days=1)
    dateList = getHKDateList(startDate=latestDate - relativedelta(months=1), endDate=latestDate, frequency='D')
    
    # Firefox Setting -> Headless
    options = Options()
    options.headless = True
    
    # Firefox Setting -> Change download path for this task
    profile = webdriver.FirefoxProfile()
    profile.set_preference("browser.download.folderList", 2)
    profile.set_preference("browser.download.manager.showWhenStarting", False)
    profile.set_preference("browser.download.dir", downloadPath)
    profile.set_preference("browser.helperApps.neverAsk.saveToDisk", "text/csv")
    
    # Link to the website
    driver = webdriver.Firefox(firefox_profile=profile, options=options)
    driver.get('https://www.hsi.com.hk/eng/indexes/all-indexes/hsi')
    timeout = 20
    try:
        element_present = EC.presence_of_element_located((By.XPATH, "//input[@aria-label='Calendar input field']"))
        WebDriverWait(driver, timeout).until(element_present)
        time.sleep(5)
    except TimeoutException:
        print("Timed out waiting for page to load")
        
    ##### Download Last Date Constituents CSV #####
    
    # Select report type in daily report
    selectReport = Select(driver.find_element_by_id('reportType'))
    selectReport.select_by_visible_text('Constituent Daily Performance (csv format)')
    
    # Input date
    latestDate = dateList[-1]
    inputDate = latestDate.strftime('%d%m%Y') 
    datePicker = driver.find_element(By.XPATH, "//input[@aria-label='Calendar input field']")
    datePicker.send_keys(inputDate)

    # Click download button
    downloadButton = driver.find_element(By.XPATH, "//button[@class='commonBtn' and @href='javascript:void(0);']")
    downloadButton.click()
    
    # Rename file - it should take a few seconds to download the file
    time.sleep(3)
    filename = max([downloadPath + "/" + f for f in os.listdir(downloadPath)],key=os.path.getctime)
    shutil.move(filename,os.path.join(downloadPath,"hsiCons.csv"))
    
    ##### Download Constituents Change File #####
    
    # Input blank date to disable constituents download button
    datePicker.send_keys('')
    
    # Select input in historical change of constituents
    selectInput = Select(driver.find_element_by_id('constituentsSelect'))
    selectInput.select_by_visible_text('Hang Seng Index and Sub-indexes')
    
    # Click View button
    viewButton = driver.find_element(By.XPATH, "//button[@class='commonBtn' and @href='javascript:void(0);']")
    viewButton.click()
    
    # Rename file - it should take a few seconds to download the file
    time.sleep(3)
    filename = max([downloadPath + "/" + f for f in os.listdir(downloadPath)],key=os.path.getctime)
    shutil.move(filename,os.path.join(downloadPath,"hsiConsChange.xlsx"))
    
    driver.quit()
    return print('Finished Downloading Raw Files for Date: ' + startDate + ' - ' + endDate)<jupyter_output><empty_output><jupyter_text>Input<jupyter_code># downloadPath = "/Users/Your Download Path"
downloadPath = "/Users/kayinman2/Desktop/Github/HS Factors/HSI Cons"
startDate = "20090930"
endDate = "20190930"
frequency = "BM"<jupyter_output><empty_output><jupyter_text>Download Raw Files from Web<jupyter_code>downloadRawFiles(downloadPath)<jupyter_output>/Users/kayinman2/Desktop/Github/HS Factors/HSI Cons
Start downloading HSI files to above path.
Finished Downloading Raw Files for Date: 20090930 - 20190930<jupyter_text>Read Raw Files and Clean Data  HK Date List<jupyter_code>hkDateList = getHKDateList(startDate=startDate, endDate=endDate, frequency=frequency)
hkDateList[:5]<jupyter_output><empty_output><jupyter_text>HSI Constituents<jupyter_code>hsiConsRaw = pd.read_csv(downloadPath + '/hsiCons.csv', low_memory=False, encoding="utf-16")
hsiConsRaw.head(3)
def reconstructHSICons(hsiConsRaw):
    consDf = hsiConsRaw.iloc[:,0].str.split("\t", expand = True)
    consDf = consDf.apply(lambda x: x.str.replace('"', ''))
    consDf = consDf.apply(lambda x: x.str.strip())
    
    consDf.columns = consDf.iloc[0,:]
    consDf = consDf.iloc[1:,:]
    
    return consDf
hsiCons = reconstructHSICons(hsiConsRaw=hsiConsRaw)
hsiCons.head(3)
def finalHSICons(hsiCons):
    finalHSICons = hsiCons[['Trade Date', 'Stock Code', 'Stock Name', 'Industry']].copy()  
    finalHSICons['Industry'] = finalHSICons['Industry'].apply(lambda x: x.split(' ')[1])
    finalHSICons['English Name'] = finalHSICons['Stock Name'].iloc[:,0]
    finalHSICons = finalHSICons.drop(columns='Stock Name')
    finalHSICons = finalHSICons.rename(columns={'English Name': 'Stock Name'})
    finalHSICons['Trade Date'] = finalHSICons['Trade Date'].astype('datetime64[ns]')
    return finalHSICons
finalHSICons = finalHSICons(hsiCons=hsiCons)
finalHSICons.head(3)
finalHSICons.info()<jupyter_output><class 'pandas.core.frame.DataFrame'>
RangeIndex: 50 entries, 1 to 50
Data columns (total 4 columns):
Trade Date    50 non-null datetime64[ns]
Stock Code    50 non-null object
Industry      50 non-null object
Stock Name    50 non-null object
dtypes: datetime64[ns](1), object(3)
memory usage: 1.7+ KB<jupyter_text>HSI Constituents Historical Change<jupyter_code>consHistChg = pd.read_excel(downloadPath + '/hsiConsChange.xlsx')
consHistChg.head(7)
def finalHistChgDf(consHistChg):
    histChgDf = consHistChg.iloc[3:,:]
    histChgDf.columns = histChgDf.iloc[0,:]
    histChgDf = histChgDf[[column for column in histChgDf.columns if 
                           'Effective Date' in column or 'Change' in column or 'Stock' in column]].copy()
    histChgDf.columns = ['Effective Date' if 'Effective Date' in column else 
                         'Change' if 'Change' in column else 
                         'Stock Code' if 'Stock Code' in column else 
                         'Stock Name' if 'Stock Name' in column else None for 
                         column in histChgDf.columns]
    histChgDf = histChgDf.iloc[2:,:]
    histChgDf = histChgDf.dropna(subset=['Stock Code'])
    histChgDf['Change'] = [1 if 'Add' in change else -1 if 'Delete' in change else np.nan 
                           for change in histChgDf['Change']]
    histChgDf['Effective Date'] = histChgDf['Effective Date'].astype('datetime64[ns]')
    histChgDf['Stock Code'] = [(str(code) + '.HK').rjust(7,str(0)) for code in histChgDf['Stock Code']]
    return histChgDf
histChgDf = finalHistChgDf(consHistChg=consHistChg)
histChgDf.head()<jupyter_output><empty_output><jupyter_text>Generate universe from Date List and Cleaned Raw Files<jupyter_code>def generateUniverse(hkDateList, finalHSICons, histChgDf):
    universeDfList = []
    latestUniverseDf = finalHSICons[['Stock Code', 'Trade Date']].copy().rename(columns={'Trade Date': 'Update Date'})
    reverseDateList = sorted(hkDateList, reverse=True)
    for date in reverseDateList:
        latestDate = pd.Timestamp(latestUniverseDf['Update Date'].drop_duplicates().values[0])
        thisPeriodChgDf = histChgDf[(histChgDf['Effective Date'] >= date) & 
                                    (histChgDf['Effective Date'] <= latestDate)].copy()
        if thisPeriodChgDf.empty == True:
            latestUniverseDf['Update Date'] = date
            thisAppendDf = latestUniverseDf.copy()
            universeDfList.append(thisAppendDf)
        else:
            dropList = [stockCode for stockCode, change in 
                        np.array(thisPeriodChgDf[['Stock Code', 'Change']]) if change == 1]
            addList = [stockCode for stockCode, change in 
                        np.array(thisPeriodChgDf[['Stock Code', 'Change']]) if change == -1]
            
            latestUniverseDf = latestUniverseDf[latestUniverseDf['Stock Code'].isin(dropList) == False].copy()
            latestUniverseDf['Update Date'] = date
            addDf = pd.DataFrame.from_dict({'Stock Code': addList})
            addDf['Update Date'] = date
            latestUniverseDf = pd.concat([latestUniverseDf, addDf])
            thisAppendDf = latestUniverseDf.copy()
            universeDfList.append(thisAppendDf)
    universeDf = pd.concat(universeDfList)
    return universeDf
universeDf = generateUniverse(hkDateList, finalHSICons, histChgDf)
universeDf.head()
universeDf.tail()
universeDf.to_csv('hsiUniverseDf.csv', index=False)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10558001617395993, 'fraction_numerical': 0.009434809955970886, 'mean_word_length': 3.828633405639913, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 31, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 43, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '65768', 'license': 'MIT', 'n_tokens_mistral': 3620, 'n_tokens_neox': 3437, 'n_words': 894, 'path': 'Scrape HSI Constituents.ipynb', 'repo_name': 'kymanj/web-scraping', 'size': '11129'}