# -*- coding: utf-8 -*- """修复 fetcher：fund_flow 单位乘数+保历史；macro 按列名映射+保历史；stock_api 全量历史+保历史""" import io ROOT = r'C:\Users\lookt\a_stock_timeline' def read(p): return io.open(p, encoding='utf-8').read() def write(p, s): io.open(p, 'w', encoding='utf-8').write(s) print('patched:', p.split('a_stock_timeline')[-1]) NL = chr(10) UP = 'from src.storage.db import upsert_df, upsert_rows, upsert_log, get_last_fetch' UP_OLD = 'from src.storage.db import upsert_df, upsert_log, get_last_fetch' # ── 2) fund_flow_fetcher ── p = ROOT + r'\src\fetcher\fund_flow_fetcher.py' s = read(p) if 'def parse_cn_amount' not in s: s = s.replace(UP_OLD, UP) helper = NL.join([ "DB = r'C:" + chr(92) + "Users" + chr(92) + "lookt" + chr(92) + "a_stock_timeline" + chr(92) + "data" + chr(92) + "a_stock.db'", "", "", "def parse_cn_amount(v):", " '''1.5亿 -> 1.5e8, -300万 -> -3e6，逗号/空白容错；无效返回 None'''", " if v is None or (isinstance(v, float) and v != v):", " return None", " s = str(v).replace(',', '').replace(' ', '').strip()", " if not s or s in ('--', '-', 'nan'):", " return None", " mult = 1.0", " if s.endswith('亿'):", " mult, s = 1e8, s[:-1]", " elif s.endswith('万'):", " mult, s = 1e4, s[:-1]", " try:", " return float(s) * mult", " except ValueError:", " return None", ""]) marker = "DB = r'C:" + chr(92) + "Users" + chr(92) + "lookt" + chr(92) + "a_stock_timeline" + chr(92) + "data" + chr(92) + "a_stock.db'" + NL s = s.replace(marker, helper, 1) old = NL.join([ " # 主力净流入转数值（去掉\"万\"/\"亿\"）", " df_ind['main_net_in'] = (df_ind['main_net_in'].astype(str)", " .str.replace(r'[\\u4e00-\\u9fa5万/亿]', '', regex=True))", " df_ind['main_net_in'] = pd.to_numeric(df_ind['main_net_in'], errors='coerce')"]) new = NL.join([ " # 主力净流入转数值：亿=1e8 万=1e4（带乘数换算）", " df_ind['main_net_in'] = df_ind['main_net_in'].map(parse_cn_amount)", " df_ind = df_ind[df_ind['main_net_in'].notna()]"]) if old in s: s = s.replace(old, new) old2 = NL.join([ " df = pd.concat(rows, ignore_index=True)", " conn = sqlite3.connect(DB)", " df.to_sql('money_flow', conn, if_exists='replace', index=False)", " conn.close()", " log.info(f'写入 money_flow: {len(df)} 行')", " return df"]) new2 = NL.join([ " df = pd.concat(rows, ignore_index=True)", " need = ['ts_code', 'trade_date', 'main_net_in',", " 'large_net_in', 'medium_net_in', 'small_net_in']", " for c in need:", " if c not in df.columns:", " df[c] = None", " n = upsert_rows(df[need], 'money_flow', conflict_cols=['ts_code', 'trade_date'])", " log.info(f'写入 money_flow: {n} 行（历史保留）')", " return df"]) if old2 in s: s = s.replace(old2, new2) write(p, s) # ── 3) macro_event_fetcher ── p = ROOT + r'\src\fetcher\macro_event_fetcher.py' s = read(p) s = s.replace(UP_OLD, UP) old = " df.columns = [c.lower() for c in df.columns]" new = NL.join([ " # 实测列（akshare 1.18.92）: 日期, 时间, 地区, 事件, 公布, 预期, 前值, 重要性", " # 按列【名称】映射，避免位置错位", " col_map = {'日期': 'event_date', '时间': 'event_time', '地区': 'source',", " '事件': 'title', '公布': 'actual', '预期': 'forecast',", " '前值': 'prev', '重要性': 'event_type'}", " df = df.rename(columns=col_map)"]) if old in s: s = s.replace(old, new, 1) old2 = NL.join([ ' df["event_type"] = df.get("importance", pd.Series([1] * len(df))).astype(int)', ' df["source"] = df.get("country", pd.Series([""] * len(df))).astype(str)']) new2 = NL.join([ ' df["event_type"] = pd.to_numeric(df["event_type"], errors="coerce").fillna(0).astype(int)', ' df["source"] = df["source"].fillna("").astype(str)']) if old2 in s: s = s.replace(old2, new2) old3 = NL.join([ 'def incremental_events(source: str = "macro_event") -> pd.DataFrame:', ' """增量采集宏观事件"""', ' # 优先 AkShare 财经日历（已知可用）', ' df = fetch_macro_calendar()', ' if df is not None and not df.empty:', ' upsert_df(df, source)', ' upsert_log(source, "", last_date=datetime.today().strftime("%Y-%m-%d"))', ' return df']) new3 = NL.join([ 'def incremental_events(source: str = "macro_event") -> pd.DataFrame:', ' """增量采集宏观事件（按 event_date+event_time+title 去重，历史保留）"""', ' df = fetch_macro_calendar()', ' if df is not None and not df.empty:', " need = [c for c in ['event_date', 'event_time', 'title', 'event_type', 'source',", " 'actual', 'forecast', 'prev', 'inserted_at'] if c in df.columns]", " upsert_rows(df[need], source,", " conflict_cols=['event_date', 'event_time', 'title'])", ' upsert_log(source, "", last_date=datetime.today().strftime("%Y-%m-%d"))', ' return df']) if old3 in s: s = s.replace(old3, new3) write(p, s) # ── 4) stock_api ── p = ROOT + r'\src\fetcher\stock_api.py' s = read(p) s = s.replace(UP_OLD, UP) old = NL.join([ " df = batch_fetch_indices(days=days)", " if df is not None and not df.empty:", ' upsert_df(df, "stock_daily")']) new = NL.join([ " df = batch_fetch_indices(days=days)", " if df is not None and not df.empty:", ' upsert_rows(df, "stock_daily", conflict_cols=["ts_code", "trade_date"])']) if old in s: s = s.replace(old, new) old_b = NL.join([ " df = fetch_daily(code, start, end)", " if df is not None and not df.empty:", ' upsert_df(df, "stock_daily")']) new_b = NL.join([ " df = fetch_daily(code, start, end)", " if df is not None and not df.empty:", ' upsert_rows(df, "stock_daily", conflict_cols=["ts_code", "trade_date"])']) if old_b in s: s = s.replace(old_b, new_b) write(p, s) print('ALL FETCHER PATCHES DONE')