starmen1265/invest_tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加投资方案、FIRE理财配置及家庭收支分析工具

Browse Source
This commit is contained in:
starmen1265
2026-07-27 22:19:48 +08:00
parent 0758b431f6
commit 04d9ca13c5
18 changed files with 6319 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/merge_family_finance.py
+473
View File
@@ -0,0 +1,473 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
家庭收支表合并脚本
读取两张招行储蓄卡流水CSV，过滤内部转账、理财调拨等，生成家庭收支明细和月度汇总。
"""
import csv
import os
import sys
from collections import defaultdict
from datetime import datetime, timedelta
# 修复Windows控制台编码
sys.stdout.reconfigure(encoding="utf-8")
DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")
FILE_1207 = os.path.join(DATA_DIR, "CMB_6214--------1207_20250726_20260726.csv")
FILE_1617 = os.path.join(DATA_DIR, "CMB_6214--------1617_20250726_20260726.csv")
OUT_DETAIL = os.path.join(DATA_DIR, "家庭收支明细.csv")
OUT_SUMMARY = os.path.join(DATA_DIR, "家庭月度汇总.csv")
OUT_LOG = os.path.join(DATA_DIR, "处理日志.txt")
def parse_cmb_csv(filepath, account):
"""解析招行CSV，返回交易列表。"""
transactions = []
with open(filepath, "r", encoding="utf-8") as f:
lines = f.readlines()
# 跳过前7行注释，从第8行(index=7)开始是表头+数据
reader = csv.reader(lines[7:])
header = next(reader) # 表头行
for fields in reader:
# 跳过空行或字段不足的行
if not fields or len(fields) < 7:
continue
date_str = fields[0].strip()
time_str = fields[1].strip()
income = fields[2].strip()
expense = fields[3].strip()
balance = fields[4].strip()
tx_type = fields[5].strip()
remark = fields[6].strip()
# 跳过空行（只有空白tab）
if not date_str:
continue
# 解析金额
amount = 0.0
direction = ""
if income and float(income) > 0:
amount = float(income)
direction = "收入"
elif expense and float(expense) > 0:
amount = float(expense)
direction = "支出"
else:
continue
# 解析日期
try:
tx_date = datetime.strptime(date_str, "%Y%m%d")
except ValueError:
continue
tx = {
"date": date_str,
"date_obj": tx_date,
"time": time_str,
"account": account,
"direction": direction,
"amount": amount,
"balance": balance,
"tx_type": tx_type,
"remark": remark,
"category": "",
"excluded": False,
"exclude_reason": "",
}
transactions.append(tx)
return transactions
def classify_transaction(tx):
"""
对单笔交易进行分类，返回 (category, excluded, exclude_reason)。
分类按优先级匹配。
"""
tx_type = tx["tx_type"]
remark = tx["remark"]
account = tx["account"]
direction = tx["direction"]
# === 1. 理财投资类（排除） ===
investment_keywords = ["理财申购", "理财赎回", "理财还本", "朝朝宝", "朝朝盈", "快速赎回", "银证转账", "银转证", "证转银", "基金赎回", "基金申购", "基金还本", "个人养老金缴存"]
for kw in investment_keywords:
if kw in tx_type:
return "理财投资", True, f"理财调拨({tx_type})"
# 银联代付-上海天天基金 = 基金赎回款，排除
if tx_type == "银联代付" and "天天基金" in remark:
return "理财投资", True, f"基金赎回款({remark})"
# 朝朝宝赎回在1617卡非常多，也需要排除
if "赎回" in tx_type and ("朝朝" in tx_type or "理财" in tx_type):
return "理财投资", True, f"理财调拨({tx_type})"
# === 2. 夫妻间内部转账（排除） ===
# 1207 → 1617 的各种转账
if account == "丈夫-1207":
if "张青" in remark:
return "内部转账", True, f"夫妻间转账({tx_type}: {remark})"
# 1207卡的微信转账（全部是给媳妇儿的）
if "微信转账" in remark and direction == "支出":
return "内部转账", True, f"夫妻间微信转账({remark})"
# 1207卡的微信转账退款
if "微信转账" in remark and direction == "收入" and tx_type == "网联退款":
return "内部转账", True, f"夫妻间微信转账退款({remark})"
# 1617卡：刘鑫(妻子)微信零钱提现回卡，属内部转账
if account == "妻子-1617" and direction == "收入" and "刘鑫" in remark:
return "内部转账", True, f"微信零钱提现({remark})"
# === 3. 信用卡还款（保留，但单独分类） ===
if "信用卡还款" in remark or "信用卡自动还款" in tx_type or "信用卡自扣" in tx_type:
return "信用卡还款", False, ""
# === 4. 退款（待配对） ===
# 携程部分退款——用户确认排除
if "退款" in tx_type and "携程" in remark:
return "退款", True, f"携程部分退款排除({remark})"
if "退款" in tx_type:
return "退款", False, "" # 暂时保留，后续配对处理
# === 5. 对外转账（1617卡微信转出，保留） ===
if account == "妻子-1617" and "微信转账" in remark and direction == "支出":
return "对外转账", False, ""
# === 6. 收入分类 ===
if direction == "收入":
# 汇入汇款-芒果光线：用户确认为之前支付宝消费退款，排除
if "汇入汇款" in tx_type and "芒果光线" in remark:
return "退款", True, f"支付宝退款-原消费退回({remark})"
# 银联代付（非理财类）单独分类
if tx_type == "银联代付":
return "银联代付收入", False, ""
# 股权结汇收益单独分类
if "结汇" in tx_type or "结汇" in remark or "股权" in remark:
return "股权结汇收益", False, ""
if tx_type == "工资" or "代发工资" in tx_type or "薪金" in remark or "EBPR" in remark or "工资" in remark or "代发" in remark:
return "工资收入", False, ""
if "保险" in remark or "理赔" in remark or "平安" in remark:
return "保险理赔", False, ""
if "公积金" in tx_type or "公积金" in remark:
return "公积金提取", False, ""
return "其他收入", False, ""
# === 7. 孩子零花钱 ===
if "支付宝-特约商户" in remark and tx["amount"] == 1.00:
return "孩子零花钱", False, ""
# === 8. 微信红包 ===
if "微信红包" in remark:
return "微信红包", False, ""
# === 9. 默认：日常消费 ===
return "日常消费", False, ""
def match_refunds(transactions):
"""
退款配对：对于退款收入，在同账户或跨账户的消费中查找同金额、相近日期(±3天)、同商户的交易。
配对成功则双双排除。
返回排除的交易集合。
"""
refunds = [tx for tx in transactions if tx["category"] == "退款" and not tx["excluded"]]
consumptions = [tx for tx in transactions if tx["direction"] == "支出" and not tx["excluded"]]
# 按金额建立消费索引
amount_index = defaultdict(list)
for i, tx in enumerate(consumptions):
amount_index[round(tx["amount"], 2)].append(i)
paired_consumption_ids = set()
paired_refund_ids = set()
for refund in refunds:
r_amount = round(refund["amount"], 2)
r_date = refund["date_obj"]
# 从备注中提取商户名
r_merchant = refund["remark"]
candidates = amount_index.get(r_amount, [])
found = False
for ci in candidates:
tx = consumptions[ci]
if id(tx) in paired_consumption_ids:
continue
# 日期在±3天内
if abs((tx["date_obj"] - r_date).days) <= 3:
# 商户名匹配（备注完全相同或包含关系）
if tx["remark"] == r_merchant or r_merchant in tx["remark"] or tx["remark"] in r_merchant:
paired_consumption_ids.add(id(tx))
paired_refund_ids.add(id(refund))
tx["excluded"] = True
tx["exclude_reason"] = f"退款配对(退款日期{refund['date']})"
refund["excluded"] = True
refund["exclude_reason"] = f"退款配对(消费日期{tx['date']})"
found = True
break
# 如果精确商户没匹配到，尝试同金额同日期范围（不同商户）
if not found:
for ci in candidates:
tx = consumptions[ci]
if id(tx) in paired_consumption_ids:
continue
if abs((tx["date_obj"] - r_date).days) <= 1:
paired_consumption_ids.add(id(tx))
paired_refund_ids.add(id(refund))
tx["excluded"] = True
tx["exclude_reason"] = f"退款配对-近似(退款日期{refund['date']})"
refund["excluded"] = True
refund["exclude_reason"] = f"退款配对-近似(消费日期{tx['date']})"
break
return len(paired_refund_ids)
def generate_detail_csv(transactions, output_path):
"""生成明细CSV。"""
kept = [tx for tx in transactions if not tx["excluded"]]
kept.sort(key=lambda x: (x["date_obj"], x["time"]))
with open(output_path, "w", encoding="utf-8-sig", newline="") as f:
writer = csv.writer(f)
writer.writerow(["日期", "时间", "账户", "收支方向", "金额", "交易类型", "交易备注", "分类"])
for tx in kept:
writer.writerow([
tx["date"],
tx["time"],
tx["account"],
tx["direction"],
f"{tx['amount']:.2f}",
tx["tx_type"],
tx["remark"],
tx["category"],
])
return len(kept)
def generate_summary_csv(transactions, output_path):
"""生成月度汇总CSV。"""
kept = [tx for tx in transactions if not tx["excluded"]]
# 按月统计
monthly = defaultdict(lambda: {
"income": 0.0, "expense": 0.0,
"categories": defaultdict(lambda: {"income": 0.0, "expense": 0.0})
})
for tx in kept:
month = tx["date"][:6] # YYYYMM
m = f"{month[:4]}-{month[4:6]}"
if tx["direction"] == "收入":
monthly[m]["income"] += tx["amount"]
monthly[m]["categories"][tx["category"]]["income"] += tx["amount"]
else:
monthly[m]["expense"] += tx["amount"]
monthly[m]["categories"][tx["category"]]["expense"] += tx["amount"]
# 收集所有出现过的分类
all_income_cats = set()
all_expense_cats = set()
for m_data in monthly.values():
for cat, vals in m_data["categories"].items():
if vals["income"] > 0:
all_income_cats.add(cat)
if vals["expense"] > 0:
all_expense_cats.add(cat)
all_income_cats = sorted(all_income_cats)
all_expense_cats = sorted(all_expense_cats)
with open(output_path, "w", encoding="utf-8-sig", newline="") as f:
writer = csv.writer(f)
# 表头
header = ["月份", "总收入", "总支出"]
for cat in all_income_cats:
header.append(f"收入-{cat}")
for cat in all_expense_cats:
header.append(f"支出-{cat}")
header.append("净结余")
writer.writerow(header)
for m in sorted(monthly.keys()):
data = monthly[m]
row = [m, f"{data['income']:.2f}", f"{data['expense']:.2f}"]
for cat in all_income_cats:
row.append(f"{data['categories'][cat]['income']:.2f}")
for cat in all_expense_cats:
row.append(f"{data['categories'][cat]['expense']:.2f}")
row.append(f"{data['income'] - data['expense']:.2f}")
writer.writerow(row)
# 合计行
total_income = sum(d["income"] for d in monthly.values())
total_expense = sum(d["expense"] for d in monthly.values())
total_row = ["合计", f"{total_income:.2f}", f"{total_expense:.2f}"]
for cat in all_income_cats:
total_row.append(f"{sum(d['categories'][cat]['income'] for d in monthly.values()):.2f}")
for cat in all_expense_cats:
total_row.append(f"{sum(d['categories'][cat]['expense'] for d in monthly.values()):.2f}")
total_row.append(f"{total_income - total_expense:.2f}")
writer.writerow(total_row)
return len(monthly)
def generate_log(transactions, excluded_stats, output_path):
"""生成处理日志。"""
lines = []
lines.append("=" * 70)
lines.append("家庭收支表处理日志")
lines.append("=" * 70)
lines.append("")
# 概况
total_1207 = sum(1 for tx in transactions if tx["account"] == "丈夫-1207")
total_1617 = sum(1 for tx in transactions if tx["account"] == "妻子-1617")
lines.append(f"原始交易总数：{len(transactions)} 笔")
lines.append(f" 丈夫-1207卡：{total_1207} 笔")
lines.append(f" 妻子-1617卡：{total_1617} 笔")
lines.append("")
# 排除统计
excluded = [tx for tx in transactions if tx["excluded"]]
lines.append(f"排除交易总数：{len(excluded)} 笔")
lines.append("-" * 50)
for cat, stats in sorted(excluded_stats.items()):
lines.append(f" {cat}: {stats['count']} 笔, 金额 ¥{stats['amount']:.2f}")
lines.append("")
# 保留统计
kept = [tx for tx in transactions if not tx["excluded"]]
lines.append(f"保留交易总数：{len(kept)} 笔")
lines.append("-" * 50)
kept_stats = defaultdict(lambda: {"count": 0, "income": 0.0, "expense": 0.0})
for tx in kept:
kept_stats[tx["category"]]["count"] += 1
if tx["direction"] == "收入":
kept_stats[tx["category"]]["income"] += tx["amount"]
else:
kept_stats[tx["category"]]["expense"] += tx["amount"]
total_income = sum(s["income"] for s in kept_stats.values())
total_expense = sum(s["expense"] for s in kept_stats.values())
for cat, stats in sorted(kept_stats.items()):
lines.append(f" {cat}: {stats['count']} 笔, 收入 ¥{stats['income']:.2f}, 支出 ¥{stats['expense']:.2f}")
lines.append("")
lines.append(f" 家庭总收入：¥{total_income:.2f}")
lines.append(f" 家庭总支出：¥{total_expense:.2f}")
lines.append(f" 家庭净结余：¥{total_income - total_expense:.2f}")
lines.append("")
# 列出排除的明细（按类别分组）
lines.append("=" * 70)
lines.append("排除交易明细")
lines.append("=" * 70)
by_cat = defaultdict(list)
for tx in excluded:
by_cat[tx["category"]].append(tx)
for cat in sorted(by_cat.keys()):
lines.append(f"\n【{cat}】({len(by_cat[cat])} 笔)")
for tx in sorted(by_cat[cat], key=lambda x: x["date_obj"]):
lines.append(f" {tx['date']} {tx['time']} | {tx['account']} | {tx['direction']} ¥{tx['amount']:.2f} | {tx['tx_type']} | {tx['remark']} | 原因: {tx['exclude_reason']}")
# 列出保留的大额交易（>5000元），供用户核查
lines.append("")
lines.append("=" * 70)
lines.append("保留的大额交易（>5000元）供核查")
lines.append("=" * 70)
big_txs = sorted([tx for tx in kept if tx["amount"] > 5000], key=lambda x: -x["amount"])
for tx in big_txs:
lines.append(f" {tx['date']} {tx['time']} | {tx['account']} | {tx['direction']} ¥{tx['amount']:.2f} | {tx['tx_type']} | {tx['remark']} | 分类: {tx['category']}")
with open(output_path, "w", encoding="utf-8") as f:
f.write("\n".join(lines))
def main():
print("=" * 60)
print("家庭收支表合并脚本")
print("=" * 60)
# 1. 解析CSV
print("\n[1] 解析CSV文件...")
txs_1207 = parse_cmb_csv(FILE_1207, "丈夫-1207")
txs_1617 = parse_cmb_csv(FILE_1617, "妻子-1617")
all_txs = txs_1207 + txs_1617
print(f" 1207卡: {len(txs_1207)} 笔")
print(f" 1617卡: {len(txs_1617)} 笔")
print(f" 合计: {len(all_txs)} 笔")
# 2. 分类
print("\n[2] 交易分类...")
for tx in all_txs:
cat, excluded, reason = classify_transaction(tx)
tx["category"] = cat
tx["excluded"] = excluded
tx["exclude_reason"] = reason
# 3. 退款配对
print("\n[3] 退款配对...")
refund_count_before = sum(1 for tx in all_txs if tx["category"] == "退款" and not tx["excluded"])
paired = match_refunds(all_txs)
print(f" 待配对退款: {refund_count_before} 笔")
print(f" 成功配对: {paired} 笔")
print(f" 未配对退款(保留为收入): {refund_count_before - paired} 笔")
# 未配对退款重新分类
for tx in all_txs:
if tx["category"] == "退款" and not tx["excluded"]:
tx["category"] = "其他收入"
# 4. 统计排除项
print("\n[4] 统计...")
excluded_stats = defaultdict(lambda: {"count": 0, "amount": 0.0})
for tx in all_txs:
if tx["excluded"]:
excluded_stats[tx["category"]]["count"] += 1
excluded_stats[tx["category"]]["amount"] += tx["amount"]
for cat, stats in sorted(excluded_stats.items()):
print(f" 排除-{cat}: {stats['count']} 笔, ¥{stats['amount']:.2f}")
kept_count = sum(1 for tx in all_txs if not tx["excluded"])
print(f" 保留: {kept_count} 笔")
# 5. 生成输出
print("\n[5] 生成输出文件...")
detail_count = generate_detail_csv(all_txs, OUT_DETAIL)
print(f" 家庭收支明细.csv: {detail_count} 笔")
summary_count = generate_summary_csv(all_txs, OUT_SUMMARY)
print(f" 家庭月度汇总.csv: {summary_count} 个月")
generate_log(all_txs, excluded_stats, OUT_LOG)
print(f" 处理日志.txt")
print("\n" + "=" * 60)
print("完成！输出文件位于 data/ 目录下。")
print("=" * 60)
if __name__ == "__main__":
main()