Python实现
import pandas as pd import numpy as np from datetime import datetime, timedelta from typing import List, Dict class DeterministicCashFlowForecaster: """确定性现金流预测器""" def __init__(self, contracts_df: pd.DataFrame, payments_df: pd.DataFrame): """ contracts_df: 合同应收款数据 - contract_id: 合同ID - amount: 应收金额 - due_date: 到期日 - client_id: 客户ID - collection_probability: 收款概率（0-1） payments_df: 应付款数据 - payment_id: 付款ID - amount: 应付金额 - due_date: 付款日 - type: 付款类型（工资/供应商/税金等） """ self.contracts = contracts_df self.payments = payments_df self.client_credit_scores = self._load_client_credit_scores() def _load_client_credit_scores(self) -> Dict: """加载客户信用评分""" # 可从CRM系统获取或本地数据库 return { 'Client_A': 0.95, # 信用良好，收款概率95% 'Client_B': 0.85, 'Client_C': 0.70, 'Default': 0.80 # 默认值 } def _adjust_collection_probability(self, client_id: str, base_prob: float) -> float: """根据客户信用调整收款概率""" credit_score = self.client_credit_scores.get(client_id, self.client_credit_scores['Default']) # 贝叶斯调整公式：P(实际收款|客户信用) = P(信用|收款) * P(收款) / P(信用) # 简化版本：加权平均 adjusted_prob = 0.7 * base_prob + 0.3 * credit_score return min(adjusted_prob, 1.0) # 不超过100% def forecast(self, start_date: datetime, end_date: datetime, freq: str = 'D') -> pd.DataFrame: """生成确定性现金流预测""" # 1. 生成日期范围 date_range = pd.date_range(start=start_date, end=end_date, freq=freq) # 2. 初始化结果DataFrame result = pd.DataFrame(index=date_range, columns=['cash_in', 'cash_out', 'net_cash']) result['cash_in'] = 0.0 result['cash_out'] = 0.0 # 3. 处理应收款 for _, contract in self.contracts.iterrows(): due_date = contract['due_date'] # 检查是否在预测期内 if start_date <= due_date <= end_date: # 调整收款概率 adj_prob = self._adjust_collection_probability( contract['client_id'], contract['collection_probability'] ) # 计算预期收款金额（考虑概率） expected_amount = contract['amount'] * adj_prob # 累加到对应日期 result.loc[due_date, 'cash_in'] += expected_amount # 4. 处理应付款 for _, payment in self.payments.iterrows(): due_date = payment['due_date'] if start_date <= due_date <= end_date: result.loc[due_date, 'cash_out'] += payment['amount'] # 5. 计算净现金流 result['net_cash'] = result['cash_in'] - result['cash_out'] # 6. 添加统计信息 result['cumulative_net_cash'] = result['net_cash'].cumsum() result['rolling_7d_avg'] = result['net_cash'].rolling(window=7, min_periods=1).mean() return result def calculate_liquidity_gaps(self, forecast_df: pd.DataFrame, min_balance: float = 1000000) -> List[Dict]: """识别现金流缺口""" # 模拟余额变化（从初始余额开始） initial_balance = 5000000 # 初始余额500万 balance = initial_balance gaps = [] for date, row in forecast_df.iterrows(): balance += row['net_cash'] # 检查是否低于最低余额要求 if balance < min_balance: gap_info = { 'date': date, 'balance': balance, 'gap_amount': min_balance - balance, 'cumulative_gap': sum(g['gap_amount'] for g in gaps) + (min_balance - balance) } gaps.append(gap_info) return gaps # 使用示例 if __name__ == "__main__": # 模拟数据 contracts_data = { 'contract_id': ['C001', 'C002', 'C003'], 'amount': [1000000, 500000, 800000], 'due_date': [ datetime(2024, 6, 15), datetime(2024, 6, 20), datetime(2024, 6, 25) ], 'client_id': ['Client_A', 'Client_B', 'Client_C'], 'collection_probability': [0.9, 0.8, 0.7] } payments_data = { 'payment_id': ['P001', 'P002', 'P003'], 'amount': [300000, 200000, 400000], 'due_date': [ datetime(2024, 6, 10), datetime(2024, 6, 15), datetime(2024, 6, 20) ], 'type': ['salary', 'supplier', 'tax'] } contracts_df = pd.DataFrame(contracts_data) payments_df = pd.DataFrame(payments_data) # 创建预测器 forecaster = DeterministicCashFlowForecaster(contracts_df, payments_df) # 生成预测 forecast = forecaster.forecast( start_date=datetime(2024, 6, 1), end_date=datetime(2024, 6, 30) ) print("确定性现金流预测:") print(forecast.head()) # 识别缺口 gaps = forecaster.calculate_liquidity_gaps(forecast, min_balance=2000000) print(f"\n现金流缺口数量: {len(gaps)}")
