(self, sid, cache_uri, freq: str = "day")
| 583 | r.tofile(str(cache_path)) |
| 584 | |
| 585 | def update(self, sid, cache_uri, freq: str = "day"): |
| 586 | cp_cache_uri = self.get_cache_dir(freq).joinpath(sid).joinpath(cache_uri) |
| 587 | meta_path = cp_cache_uri.with_suffix(".meta") |
| 588 | if not self.check_cache_exists(cp_cache_uri, suffix_list=[".meta"]): |
| 589 | self.logger.info(f"The cache {cp_cache_uri} has corrupted. It will be removed") |
| 590 | self.clear_cache(cp_cache_uri) |
| 591 | return 2 |
| 592 | |
| 593 | with CacheUtils.writer_lock(self.r, f"{str(C.dpm.get_data_uri())}:expression-{cache_uri}"): |
| 594 | with meta_path.open("rb") as f: |
| 595 | d = pickle.load(f) |
| 596 | instrument = d["info"]["instrument"] |
| 597 | field = d["info"]["field"] |
| 598 | freq = d["info"]["freq"] |
| 599 | last_update_time = d["info"]["last_update"] |
| 600 | |
| 601 | # get newest calendar |
| 602 | from .data import Cal, ExpressionD # pylint: disable=C0415 |
| 603 | |
| 604 | whole_calendar = Cal.calendar(start_time=None, end_time=None, freq=freq) |
| 605 | # calendar since last updated. |
| 606 | new_calendar = Cal.calendar(start_time=last_update_time, end_time=None, freq=freq) |
| 607 | |
| 608 | # get append data |
| 609 | if len(new_calendar) <= 1: |
| 610 | # Including last updated calendar, we only get 1 item. |
| 611 | # No future updating is needed. |
| 612 | return 1 |
| 613 | else: |
| 614 | # get the data needed after the historical data are removed. |
| 615 | # The start index of new data |
| 616 | current_index = len(whole_calendar) - len(new_calendar) + 1 |
| 617 | |
| 618 | # The existing data length |
| 619 | size_bytes = os.path.getsize(cp_cache_uri) |
| 620 | ele_size = np.dtype("<f").itemsize |
| 621 | assert size_bytes % ele_size == 0 |
| 622 | ele_n = size_bytes // ele_size - 1 |
| 623 | |
| 624 | expr = ExpressionD.get_expression_instance(field) |
| 625 | lft_etd, rght_etd = expr.get_extended_window_size() |
| 626 | # The expression used the future data after rght_etd days. |
| 627 | # So the last rght_etd data should be removed. |
| 628 | # There are most `ele_n` period of data can be remove |
| 629 | remove_n = min(rght_etd, ele_n) |
| 630 | assert new_calendar[1] == whole_calendar[current_index] |
| 631 | data = self.provider.expression( |
| 632 | instrument, field, whole_calendar[current_index - remove_n], new_calendar[-1], freq |
| 633 | ) |
| 634 | with open(cp_cache_uri, "ab") as f: |
| 635 | data = np.array(data).astype("<f") |
| 636 | # Remove the last bits |
| 637 | f.truncate(size_bytes - ele_size * remove_n) |
| 638 | f.write(data) |
| 639 | # update meta file |
| 640 | d["info"]["last_update"] = str(new_calendar[-1]) |
| 641 | with meta_path.open("wb") as f: |
| 642 | pickle.dump(d, f, protocol=C.dump_protocol_version) |
nothing calls this directly
no test coverage detected