# -*- coding: utf-8 -*- r""" Market_Wave 用：inside_log_*.txt → Excel集計 Ver4 使い方： cd C:\Users\machi\Market_Wave\05_ツール # 1日だけ作成 python .\text_to_inside_excel_wave_v4.py "2026-07-02" # 00_波テキスト内の inside_log_*.txt を全部まとめる python .\text_to_inside_excel_wave_v4.py 保存先： C:\Users\machi\Market_Wave\01_波Excel Ver4の特徴： - スクリプトの置き場所から Market_Wave フォルダを自動判定 - 1分波を読み取り - 30分中身がログに無い場合、1分波から30分中身を自動生成 - all_inside_radar_summary.xlsx に全日分をまとめる - 日別・銘柄別・時間帯別・ランキングを作成 """ import re import sys from pathlib import Path from typing import Optional, List, Tuple import pandas as pd # ============================================================ # フォルダ設定：C:\Users\machi\Market_Wave\05_ツール に置けば自動で親を取得 # ============================================================ SCRIPT_DIR = Path(__file__).resolve().parent BASE = SCRIPT_DIR.parent TEXT_DIR = BASE / "00_波テキスト" OUT_DIR = BASE / "01_波Excel" ANALYSIS_DIR = BASE / "02_解析" OUT_DIR.mkdir(parents=True, exist_ok=True) ANALYSIS_DIR.mkdir(parents=True, exist_ok=True) ZEN2HAN = str.maketrans("０１２３４５６７８９", "0123456789") # 30分区間。東証の昼休みを考慮。15:00-15:30も入れる。 HALF_HOUR_BINS = [ ("09:00", "09:30"), ("09:30", "10:00"), ("10:00", "10:30"), ("10:30", "11:00"), ("11:00", "11:30"), ("12:30", "13:00"), ("13:00", "13:30"), ("13:30", "14:00"), ("14:00", "14:30"), ("14:30", "15:00"), ("15:00", "15:30"), ] def normalize_text(s: str) -> str: return s.translate(ZEN2HAN).replace("　", " ") def extract_date(text: str) -> Optional[str]: text = normalize_text(text) m = re.search(r"\d{4}-\d{2}-\d{2}", text) return m.group(0) if m else None def pick_txt_files(arg: Optional[str]) -> Tuple[Optional[str], List[Path]]: """ 1) 引数なし -> 00_波テキスト内 inside_log_*.txt 全部 2) 日付指定 -> inside_log_日付.txt だけ 3) フルパス -> その1ファイルだけ """ if not arg: return None, sorted(TEXT_DIR.glob("inside_log_*.txt")) arg_norm = normalize_text(arg) p = Path(arg) if p.exists() and p.is_file() and p.suffix.lower() == ".txt": target_date = extract_date(p.name) or p.stem return target_date, [p] target_date = extract_date(arg_norm) or arg_norm txt_files = sorted(TEXT_DIR.glob(f"inside_log_{target_date}.txt")) return target_date, txt_files def safe_float(x): return float(str(x).replace(",", "")) def safe_int(x): return int(float(str(x).replace(",", ""))) def parse_1min_wave(txt_file: Path, text: str) -> List[dict]: rows = [] current_name = None current_date = None in_table = False for line in text.splitlines(): raw = normalize_text(line).strip() if not raw: continue # 例：7013_IHI 2026-07-02 1分波観測一覧 m = re.search(r"([0-9]{3,4}[A-Z]?_[^\s]+)\s+(\d{4}-\d{2}-\d{2})\s+1分波観測一覧", raw) if m: current_name = m.group(1).strip() current_date = m.group(2) in_table = True continue if "日別まとめ" in raw or "====" in raw and in_table: # 見出し線は無視。日別まとめに入ったら終了。 if "日別まとめ" in raw: in_table = False continue # 例：09:00-09:01 1906.0 1883.0 92400 1.03 強い下落 第1波_下げ波 戻り売り確認 if in_table and current_name and re.match(r"^\d{2}:\d{2}-\d{2}:\d{2}\s+", raw): parts = raw.split() if len(parts) < 6: continue try: rows.append({ "元ファイル": txt_file.name, "日付": current_date, "銘柄": current_name, "区間": parts[0], "始値": safe_float(parts[1]), "終値": safe_float(parts[2]), "出来高": safe_int(parts[3]), "出来高倍率": safe_float(parts[4]), "方向": parts[5], "波判定": parts[6] if len(parts) >= 7 else "", "罠判定": " ".join(parts[7:]) if len(parts) >= 8 else "", }) except Exception as e: print("1分波読み取り失敗:", current_name, raw, e) return rows def parse_30min_inside_old_style(txt_file: Path, text: str) -> List[dict]: """旧ログ形式に30分足の中身透視レーダーが残っている場合だけ読む。""" rows = [] blocks = re.split(r"\n(?=[0-9]{3,4}[A-Z]?_.+?30分足の中身透視レーダー 開始)", text) for block in blocks: m = re.search(r"([0-9]{3,4}[A-Z]?_.+?)\s*30分足の中身透視レーダー 開始", block) if not m: continue name = m.group(1).strip() file_date = extract_date(txt_file.name) for line in block.splitlines(): raw = normalize_text(line).strip() if re.match(r"^\d+\s+\d{2}:\d{2}-\d{2}:\d{2}", raw): parts = raw.split() if len(parts) < 9: continue try: rows.append({ "元ファイル": txt_file.name, "日付": file_date, "銘柄": name, "区間": parts[1], "始値": safe_float(parts[2]), "高値": safe_float(parts[3]), "安値": safe_float(parts[4]), "終値": safe_float(parts[5]), "出来高": safe_int(parts[6]), "30分方向": parts[7], "波の数": safe_int(parts[8]), "中身判定": " ".join(parts[9:]), "作成方法": "ログ直接読取", }) except Exception as e: print("30分読み取り失敗:", name, raw, e) return rows def time_to_minutes(t: str) -> int: h, m = t.split(":") return int(h) * 60 + int(m) def bin_30min(interval: str) -> Optional[str]: start = interval.split("-")[0] minute = time_to_minutes(start) for s, e in HALF_HOUR_BINS: if time_to_minutes(s) <= minute < time_to_minutes(e): return f"{s}-{e}" return None def direction_from_change(open_p: float, close_p: float) -> str: if open_p == 0: return "横ばい" pct = (close_p - open_p) / open_p * 100 if pct >= 0.8: return "強い上昇" if pct >= 0.15: return "上昇寄り" if pct <= -0.8: return "強い下落" if pct <= -0.15: return "下落寄り" return "横ばい" def judge_inside(up_count: int, down_count: int, direction: str) -> str: if up_count > down_count and "上昇" in direction: return "上昇で中身も買い優勢" if down_count > up_count and "下落" in direction: return "下落で中身も売り優勢" if up_count > down_count and "下落" in direction: return "表面は下落・中身に反発あり" if down_count > up_count and "上昇" in direction: return "表面は上昇・中身に戻り売りあり" if up_count == down_count and (up_count + down_count) > 0: return "表面は横ばい・中身は乱戦" return "中立" def make_30min_from_1min(dfwave: pd.DataFrame) -> pd.DataFrame: if dfwave.empty: return pd.DataFrame() df = dfwave.copy() df["30分区間"] = df["区間"].map(bin_30min) df = df.dropna(subset=["30分区間"]) if df.empty: return pd.DataFrame() rows = [] for (file_name, date, name, half), g in df.groupby(["元ファイル", "日付", "銘柄", "30分区間"], sort=True): g = g.reset_index(drop=True) open_p = float(g.loc[0, "始値"]) close_p = float(g.loc[len(g) - 1, "終値"]) high_p = float(pd.concat([g["始値"], g["終値"]]).max()) low_p = float(pd.concat([g["始値"], g["終値"]]).min()) volume = int(g["出来高"].sum()) direction = direction_from_change(open_p, close_p) up_count = int(g["波判定"].astype(str).str.contains("上げ波", na=False).sum()) down_count = int(g["波判定"].astype(str).str.contains("下げ波", na=False).sum()) wave_count = up_count + down_count max_ratio = float(g["出来高倍率"].max()) if "出来高倍率" in g else 0.0 pct = ((close_p - open_p) / open_p * 100) if open_p else 0.0 rows.append({ "元ファイル": file_name, "日付": date, "銘柄": name, "区間": half, "始値": open_p, "高値": high_p, "安値": low_p, "終値": close_p, "騰落率%": round(pct, 3), "出来高": volume, "最大出来高倍率": round(max_ratio, 2), "30分方向": direction, "波の数": wave_count, "上げ波回数": up_count, "下げ波回数": down_count, "中身判定": judge_inside(up_count, down_count, direction), "作成方法": "1分波から自動生成", }) return pd.DataFrame(rows) def add_ranking_sheets(writer, dfwave: pd.DataFrame, df30: pd.DataFrame): if not dfwave.empty: wave_summary = ( dfwave.groupby(["日付", "銘柄"], as_index=False) .agg( 総出来高=("出来高", "sum"), 最大出来高=("出来高", "max"), 最大出来高倍率=("出来高倍率", "max"), 記録行数=("区間", "count"), 上げ波回数=("波判定", lambda s: s.astype(str).str.contains("上げ波", na=False).sum()), 下げ波回数=("波判定", lambda s: s.astype(str).str.contains("下げ波", na=False).sum()), 強い上昇回数=("方向", lambda s: s.astype(str).str.contains("強い上昇", na=False).sum()), 強い下落回数=("方向", lambda s: s.astype(str).str.contains("強い下落", na=False).sum()), ) ) wave_summary["上げ波率%"] = ((wave_summary["上げ波回数"] / wave_summary["記録行数"]) * 100).round(1) wave_summary["下げ波率%"] = ((wave_summary["下げ波回数"] / wave_summary["記録行数"]) * 100).round(1) wave_summary.to_excel(writer, index=False, sheet_name="1分波_銘柄別summary") wave_summary.sort_values("総出来高", ascending=False).head(100).to_excel(writer, index=False, sheet_name="出来高ランキング") wave_summary.sort_values("最大出来高倍率", ascending=False).head(100).to_excel(writer, index=False, sheet_name="倍率ランキング") wave_summary.sort_values("下げ波回数", ascending=False).head(100).to_excel(writer, index=False, sheet_name="下げ波ランキング") wave_summary.sort_values("上げ波回数", ascending=False).head(100).to_excel(writer, index=False, sheet_name="上げ波ランキング") if not df30.empty: summary30 = ( df30.groupby(["日付", "銘柄"], as_index=False) .agg( 総出来高=("出来高", "sum"), 総波数=("波の数", "sum"), 平均波数=("波の数", "mean"), 区間数=("区間", "count"), 最大30分出来高=("出来高", "max"), ) ) summary30["平均波数"] = summary30["平均波数"].round(2) summary30.to_excel(writer, index=False, sheet_name="30分_銘柄別summary") time_summary = ( df30.groupby("区間", as_index=False) .agg( 銘柄数=("銘柄", "nunique"), 合計出来高=("出来高", "sum"), 平均波数=("波の数", "mean"), 上昇寄り数=("30分方向", lambda s: s.astype(str).str.contains("上昇", na=False).sum()), 下落寄り数=("30分方向", lambda s: s.astype(str).str.contains("下落", na=False).sum()), ) ) time_summary["平均波数"] = time_summary["平均波数"].round(2) time_summary.to_excel(writer, index=False, sheet_name="時間帯summary") # 銘柄×時間帯の出来高ピボット。Market Waveの“波地図”。 pivot_vol = df30.pivot_table(index="銘柄", columns="区間", values="出来高", aggfunc="sum", fill_value=0) pivot_vol.reset_index().to_excel(writer, index=False, sheet_name="銘柄別_時間帯出来高") def autosize_excel(path: Path): try: from openpyxl import load_workbook from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter from openpyxl.formatting.rule import CellIsRule wb = load_workbook(path) header_fill = PatternFill("solid", fgColor="1F4E78") header_font = Font(color="FFFFFF", bold=True) thin = Side(style="thin", color="D9E2F3") border = Border(left=thin, right=thin, top=thin, bottom=thin) for ws in wb.worksheets: ws.freeze_panes = "A2" ws.auto_filter.ref = ws.dimensions for cell in ws[1]: cell.fill = header_fill cell.font = header_font cell.alignment = Alignment(horizontal="center", vertical="center") cell.border = border for row in ws.iter_rows(min_row=2): for cell in row: cell.border = border cell.alignment = Alignment(vertical="center") for col in ws.columns: col_letter = get_column_letter(col[0].column) max_len = max(len(str(c.value)) if c.value is not None else 0 for c in col) ws.column_dimensions[col_letter].width = min(max(max_len + 2, 10), 32) wb.save(path) except Exception as e: print("Excel整形はスキップしました:", e) def main(): arg = sys.argv[1] if len(sys.argv) >= 2 else None target_date, txt_files = pick_txt_files(arg) if not txt_files: print("対象ファイルが見つかりません") print(f"指定: {arg}") print(f"検索場所: {TEXT_DIR}") sys.exit(1) rows_30_old = [] rows_wave = [] for txt_file in txt_files: text = txt_file.read_text(encoding="utf-8", errors="ignore") text = normalize_text(text) rows_30_old.extend(parse_30min_inside_old_style(txt_file, text)) rows_wave.extend(parse_1min_wave(txt_file, text)) dfwave = pd.DataFrame(rows_wave) df30_old = pd.DataFrame(rows_30_old) df30_auto = make_30min_from_1min(dfwave) # 旧30分があれば採用。ただし無ければ1分から作る。 # 両方ある場合は、旧30分を優先しつつ、自動生成も参考として残す。 if not df30_old.empty: df30_main = df30_old.copy() else: df30_main = df30_auto.copy() if target_date: out_file = OUT_DIR / f"all_inside_radar_summary_{target_date}.xlsx" else: out_file = OUT_DIR / "all_inside_radar_summary.xlsx" with pd.ExcelWriter(out_file, engine="openpyxl") as writer: if not df30_main.empty: df30_main.to_excel(writer, index=False, sheet_name="全銘柄_30分中身") if not dfwave.empty: dfwave.to_excel(writer, index=False, sheet_name="全銘柄_1分波") if not df30_auto.empty: df30_auto.to_excel(writer, index=False, sheet_name="30分_自動生成確認") add_ranking_sheets(writer, dfwave, df30_main) if df30_main.empty and dfwave.empty: pd.DataFrame([{ "メッセージ": "読み取れる行がありませんでした", "確認": "入力ログの形式が想定外か、表部分が貼り付けられていない可能性があります。", }]).to_excel(writer, index=False, sheet_name="読み取り結果") autosize_excel(out_file) print("保存完了") print(out_file) print(f"処理ファイル数: {len(txt_files)}") print(f"30分中身 行数: {len(df30_main)}") print(f" - ログ直接読取: {len(df30_old)}") print(f" - 1分波から自動生成: {len(df30_auto)}") print(f"1分波 行数: {len(dfwave)}") for f in txt_files: print(f"処理: {f.name}") if __name__ == "__main__": main()