(self)
| 224 | return pd.Timestamp("1999-01-01") |
| 225 | |
| 226 | def get_changes(self) -> pd.DataFrame: |
| 227 | logger.info(f"get sp500 history changes......") |
| 228 | # NOTE: may update the index of the table |
| 229 | changes_df = pd.read_html(self.WIKISP500_CHANGES_URL)[-1] |
| 230 | changes_df = changes_df.iloc[:, [0, 1, 3]] |
| 231 | changes_df.columns = [self.DATE_FIELD_NAME, self.ADD, self.REMOVE] |
| 232 | changes_df[self.DATE_FIELD_NAME] = pd.to_datetime(changes_df[self.DATE_FIELD_NAME]) |
| 233 | _result = [] |
| 234 | for _type in [self.ADD, self.REMOVE]: |
| 235 | _df = changes_df.copy() |
| 236 | _df[self.CHANGE_TYPE_FIELD] = _type |
| 237 | _df[self.SYMBOL_FIELD_NAME] = _df[_type] |
| 238 | _df.dropna(subset=[self.SYMBOL_FIELD_NAME], inplace=True) |
| 239 | if _type == self.ADD: |
| 240 | _df[self.DATE_FIELD_NAME] = _df[self.DATE_FIELD_NAME].apply( |
| 241 | lambda x: get_trading_date_by_shift(self.calendar_list, x, 0) |
| 242 | ) |
| 243 | else: |
| 244 | _df[self.DATE_FIELD_NAME] = _df[self.DATE_FIELD_NAME].apply( |
| 245 | lambda x: get_trading_date_by_shift(self.calendar_list, x, -1) |
| 246 | ) |
| 247 | _result.append(_df[[self.DATE_FIELD_NAME, self.CHANGE_TYPE_FIELD, self.SYMBOL_FIELD_NAME]]) |
| 248 | logger.info(f"end of get sp500 history changes.") |
| 249 | return pd.concat(_result, sort=False) |
| 250 | |
| 251 | def filter_df(self, df: pd.DataFrame) -> pd.DataFrame: |
| 252 | if "Symbol" in df.columns: |
nothing calls this directly
no test coverage detected