import logging from SQLiteInterface import SQLiteInterface # 定义一个装饰器函数，用于捕捉和处理指定的异常 def exception_handler(expected_exception=Exception, error_msg="An error occurred"): def decorator(func): def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except expected_exception as e: logging.error(f"{error_msg}: {e}") return None return wrapper return decorator # 定义现金流标签引擎类 class CashFlowLabelEngine: def __init__(self, db_interface: SQLiteInterface, target=None): self.db_interface = db_interface self.target_table = target # 主程序 @exception_handler(expected_exception=Exception, error_msg="Error processing transactions") def execute(self): # 查询未标记的交易记录 query = f"SELECT * FROM {self.target_table} WHERE label IS NULL;" transactions = self.db_interface.read(query) if not transactions: logging.error("No transactions found or failed to read from the database.") return updates = [] for transaction in transactions: # 构建交易字典并标记单笔交易 trans_dict = self.construct_transaction_dict(transaction) label = self.label_single_transaction(trans_dict) updates.append((label, trans_dict['cashflow_id'])) # 批量更新数据库中的交易标签 update_query = f"UPDATE {self.target_table} SET label = ? WHERE cashflow_id = ?;" self.db_interface.batch_write(update_query, updates) # 构建交易字典 @exception_handler(expected_exception=IndexError, error_msg="Error constructing transaction dictionary") def construct_transaction_dict(self, transaction): trans_dict = { 'cashflow_id': transaction[0], 'category': transaction[2], 'counterparty': transaction[3], 'description': transaction[4], 'transaction_type': transaction[5], 'amount': transaction[6], 'payment_method': transaction[7], 'status': transaction[8], 'note': transaction[9], 'cashflow_account_name': transaction[12] } return trans_dict # 标记单笔交易 @exception_handler(expected_exception=Exception, error_msg="Error in labeling single transaction") def label_single_transaction(self, transaction): operating_label = self.label_operating(transaction) if operating_label != 'Unknown': return operating_label investing_label = self.label_investing(transaction) if investing_label != 'Unknown': return investing_label financing_label = self.label_financing(transaction) if financing_label != 'Unknown': return financing_label return 'Unknown' # 匹配交易条件 @exception_handler(expected_exception=TypeError, error_msg="Error matching transaction criteria") def matches_criteria(self, transaction, criteria, logic='AND'): results = [] for field, (operation, value) in criteria.items(): transaction_value = transaction.get(field, '') if transaction_value is None: # 检查字段值是否为 None transaction_value = '' # 将 None 转换为空字符串或其他适当的默认值 if operation == 'contains': result = any(keyword in transaction_value for keyword in value) elif operation == 'not_contains': result = all(keyword not in transaction_value for keyword in value) elif operation == 'less_than': result = float(transaction_value) < value elif operation == 'greater_than': result = float(transaction_value) > value elif operation == 'equals': result = float(transaction_value) == value else: result = False results.append(result) if logic == 'AND': return all(results) elif logic == 'OR': return any(results) elif logic == 'NOT': return not any(results) return False # 运营活动标注方法 @exception_handler(expected_exception=Exception, error_msg="Error in operating label logic") def label_operating(self, transaction): if self.is_salary_income(transaction): return 'SalaryIncome' elif self.is_good_income(transaction): return 'GoodIncome' elif self.is_other_income(transaction): return 'OtherIncome' elif self.is_none_income(transaction): return 'NoneIncome' elif self.is_none_income_expense(transaction): return 'NoneIncomeExpenses' elif self.is_fixed_expenses(transaction): return 'FixedExpenses' elif self.is_daily_expense(transaction): return 'DailyExpenses' elif self.is_personalized_expense(transaction): return 'PersonalizedExpenses' else: return 'Unknown' # 判断是否为工资收入 def is_salary_income(self, transaction): criteria = { 'transaction_type': ('contains', ['收入']), 'category': ('contains', ['收入']), 'description': ('contains', ['红包']) } return self.matches_criteria(transaction, criteria, logic='AND') # 判断是否为商品收入 def is_good_income(self, transaction): criteria = { 'transaction_type': ('contains', ['收入']), 'note': ('contains', ['赞赏作者']), 'cashflow_account_name': ('contains', 'weichatpay') } return self.matches_criteria(transaction, criteria, logic='AND') # 判断是否为其他收入 def is_other_income(self, transaction): positive_criteria = { 'transaction_type': ('contains', ['收入']), 'counterparty': ('contains', ['爸爸', '妈妈']) } exclusion_criteria = { 'counterparty': ('not_contains', ['老婆', '四川电信']) } if not self.matches_criteria(transaction, exclusion_criteria, logic='AND'): return False result = self.matches_criteria(transaction, positive_criteria, logic='AND') return result # 判断是否为非收入 def is_none_income(self, transaction): criteria1 = { 'transaction_type': ('not_contains', ['支出']), 'status': ('contains', ['退款成功', '交易成功']), 'category': ('contains', ['退款', '保险']), } criteria2 = { 'transaction_type': ('not_contains', ['支出']), 'description': ('contains', ['退款', '退保', 'iPhone']), } return self.matches_criteria(transaction, criteria1, logic='AND') or self.matches_criteria(transaction, criteria2, logic='AND') # 判断是否为非收入支出 def is_none_income_expense(self, transaction): criteria = { 'status': ('contains', ['退款成功', '已全额退款', '交易关闭']), 'cashflow_account_name': ('contains', ['alipay', 'weichatpay']) } return self.matches_criteria(transaction, criteria, logic='AND') # 判断是否为固定支出 def is_fixed_expenses(self, transaction): if transaction['transaction_type'] != '支出': return False exclusion_criteria = { 'description': ('contains', ['CGV', '芒果TV']), 'counterparty': ('contains', ['成都天府通数字科技有限公司']) } if self.matches_criteria(transaction, exclusion_criteria, logic='OR'): return False positive1_criteria = { 'category': ('contains', ['充值缴费', '养老保险', '保险', '家居家装', '医疗健康']), 'description': ('contains', ['墨盒', '话费', '四川电信', 'sms', '保险', '生活缴费', '美年大健康', '医院', '物业', '电费', '水费', '燃气费', '沁园']), 'counterparty': ('contains', ['社保', '四川电信', '医院', 'iCloud', '四川省税务局', '保险']) } positive2_criteria = { 'counterparty': ('contains', ['成都易智禧科技有限公司', '小艾', '苹果电子产品商贸(北京)有限公司', '小米科技有限责任公司']), 'amount': ('less_than', 500) } if not (self.matches_criteria(transaction, positive1_criteria, logic='OR') or self.matches_criteria(transaction, positive2_criteria, logic='AND')): return False return True # 判断是否为日常支出 def is_daily_expense(self, transaction): positive_criteria = { 'transaction_type': ('contains', ['支出']), 'category': ('contains', ['餐饮美食', '扫二维码付款', '商户消费', '交通出行', '信用借还']), 'counterparty': ('contains', ['廖记', '瓜子炒货', '钱大妈', '吴记喜利来老面馆', '叁顿大包', '伊藤洋华堂', '成都天府通数字科技有限公司', '盒马', '袁记', '成都金控数据服务有限公司', '小米移动', '小米有品', '公交', '电车', '盒马', '收款方备注', '单车', '成都地铁', '充电宝', '停车', '超市', '车费', '储值', '乐山钵钵鸡']), 'amount': ('less_than', 400) } exclusion_criteria = { 'description': ('not_contains', ['星巴克', 'Arabica', '哈根达斯', '曹操出行', 'luckincoffee', '蜜雪冰城']), 'counterparty': ('not_contains', ['咖啡', '高挺', '支付宝']) } if not self.matches_criteria(transaction, exclusion_criteria, logic='AND'): return False if not (self.matches_criteria(transaction, positive_criteria, logic='AND')): return False return True # 判断是否为个性化支出 def is_personalized_expense(self, transaction): criteria = { 'transaction_type': ('contains', ['支出']), 'category': ('contains', ['数码电器', '亲友代付', '亲属卡交易', '公益捐赠', '酒店旅游', '微信红包', '商户消费', '文化休闲', '生活服务', '转账', '商业服务']), 'description': ('contains', ['机票', '火车票', '收钱码收款', '滴滴', '曹操', '星巴克', '喜茶', '美团', 'CGV', '拿铁', 'App Store & Apple Music', '月卡', '芒果TV']), 'counterparty': ('contains', ['支付宝红包', '中国社会福利基金会', '淘票', '中国铁路网络有限公司', '茶百道', '蜜雪冰城', 'luckincoffee', 'M Stand', '咖啡']), 'description': ('not_contains', ['信用卡还款', '朝朝宝转入', '基金申购', '生活缴费', 'apple.com', '挂号订单', '话费', '税务局']), 'counterparty': ('not_contains', ['基金管理', '基金销售', '保险', '理财', '证券', '还款', '网商', '苹果电子产品商贸(北京)有限公司', '小米', '京东']) } return self.matches_criteria(transaction, criteria, logic='AND') # 投资活动标注方法 @exception_handler(expected_exception=Exception, error_msg="Error in investing label logic") def label_investing(self, transaction): if self.is_investment_liquidation(transaction): return 'InvestmentLiquidation' elif self.is_investment_returns(transaction): return 'InvestmentReturns' elif self.is_cashflow_from_capex(transaction): return 'CashflowFromCapEx' elif self.is_capital_expenditures(transaction): return 'CapitalExpenditures' elif self.is_pp_portfolio(transaction): return 'PerpetualPortfolio' else: return 'Unknown' # 判断是否为投资清算 def is_investment_liquidation(self, transaction): criteria = { 'transaction_type': ('not_contains', ['支出']), 'category': ('contains', ['投资理财']), 'description': ('contains', ['卖出']) } return self.matches_criteria(transaction, criteria, logic='AND') # 判断是否为投资回报 def is_investment_returns(self, transaction): criteria = { 'transaction_type': ('not_contains', ['支出']), 'category': ('contains', ['投资理财', '利息收入']), 'description': ('contains', ['收益发放', '现金分红', '账户结息']) } return self.matches_criteria(transaction, criteria, logic='AND') # 判断是否为资本支出现金流 def is_cashflow_from_capex(self, transaction): criteria = { 'transaction_type': ('not_contains', ['支出']), 'category': ('contains', ['保险', '转账']), 'description': ('contains', ['保险退保', '汇入汇款', '转账']), 'counterparty': ('contains', ['中国人民人寿保险股份有限公司']) } return self.matches_criteria(transaction, criteria, logic='AND') # 判断是否为资本支出 def is_capital_expenditures(self, transaction): if transaction['transaction_type'] != '支出': return False criteria = { 'amount': ('less_than', 100) # 微信的资本支出通常是金额较小的转账 } if self.matches_criteria(transaction, criteria, logic='AND'): return False criteria1 = { 'description': ('contains', ['京东', '小米']), 'counterparty': ('contains', ['成都易智禧科技有限公司', '小艾', '苹果电子产品商贸(北京)有限公司', '小米', '京东']), } return self.matches_criteria(transaction, criteria1, logic='OR') # 判断是否为永久投资组合 def is_pp_portfolio(self, transaction): criteria = { 'description': ('contains', ['华安纳斯达克', '博时标普500', '余额＋', '支付宝转个人养老金资金账户']), 'category': ('contains', ['购买理财通']) # 假设包含购买理财通的类别 } return self.matches_criteria(transaction, criteria, logic='AND') # 融资活动标注方法 @exception_handler(expected_exception=Exception, error_msg="Error in financing label logic") def label_financing(self, transaction): if self.is_mortgage_payments(transaction): return 'MortgagePayments' elif self.is_credit_payments(transaction): return 'LivingPayments' # 其他融资活动标签的判断逻辑可以继续添加 else: return 'Unknown' # 判断是否为按揭还款 def is_mortgage_payments(self, transaction): criteria = { 'transaction_type': ('not_contains', ['收入']), 'counterparty': ('contains', ['中国工商银行']) } return self.matches_criteria(transaction, criteria, logic='AND') # 判断是否为信用支付 def is_credit_payments(self, transaction): criteria = { 'transaction_type': ('not_contains', ['收入']), 'category': ('contains', ['信用借还', '信用卡还款']), 'description': ('contains', ['自动还款','还款', '信用卡', '/', '快捷支付']), } return self.matches_criteria(transaction, criteria, logic='AND') # 执行程序 db_interface = SQLiteInterface() labeler = CashFlowLabelEngine(db_interface=db_interface, target='WeichatpayCashFlowStatement') labeler.execute()