资金流入流出预测—时间序列规则

2024-01-24 02:58
文章标签 时间 规则 预测 序列 资金 流出 流入

本文主要是介绍资金流入流出预测—时间序列规则，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

资金流入流出预测—时间序列规则

选择特征

可以用简单的统计量来作为特征，从中提取出有用的信息

  • 中位数：按顺序排列的一组数据中居于中间位置的数，较为稳健
  • 均值：当分布较符合正态分布时，可以较好的代表整体特征
  • 临近数据：离待测数据越近的数据对其影响越大

基于周期因子的时间序列预测

支付数据、客流量数据、交通数据等时间序列通常都具有明显的周期性。当判断出时间序列的周期性后，就需要确定周期的长度（一周或一个月）和组成一个周期的元素（周一到周日、1日至31日），然后结合STL分解 (Seasonal and Trend decomposition using Loess) 观察周期变化。但是上述过程没有考虑到节假日、突发事件等情况造成的影响。

基本规则法（baseline）

假设给定下面的数据，任务是根据前三周的数据预测第四周每天的客流量。

周一周二周三周四周五周六周日周均值
第一周20107050250200100100
第二周261866501801408080
第三周1586760270160120100

在这里插入图片描述
从图中可以明显地看到周一到周日的周期波动。预测的核心任务就是尽可能准确的提取这种周期。

STEP 1：获得周期因子

  • 方式1：每天的数据除以周均值 ，然后按列取中位数

    在这里插入图片描述

  • 方式2：季节指数的计算方式，获得每日（星期几）均值，再除以整体均值

在这里插入图片描述
STEP 2：获得base

做预测时，只要将周期因子乘以base，就可以得到下一周的预测。例如，取最后一周的平均客流量100作为base，得到如下结果：

在这里插入图片描述
预测下个月每一天的情况：

如果想预测下个月每天的客流量情况，需要提取以月为周期的因子，具体可以采用如下方式：

  • 获得每日（1号-31号）均值
  • 统计每日（周一至周日）的频次
  • 基于星期周期因子获得加权均值
  • 根据因子和每日均值进行预测

针对周期因子的优化

按列提取中位数是一种简单而有效的提取周期因子的方法。中位数具有鲁棒性，不受极端值的影响。但中位数损失了很多信息。在实践中，可以在此基础上进一步优化，比如可以提取一个均值和一个中位数，然后将均值和中位数融合。融合的比例按照测试集的表现来确定。也可以根据与预测周的时间距离来赋予不同的权重。

针对base的优化

直接用最后一周的平均客流量作为base并不一定是最好的方法，也许最后三天或最后五天的均值能更好的反映最新的情况。但是，直接对最后三天客流量取均值的做法并不合适（最后三天是周末，这样取的base会偏大）。需要去掉周期性因素后，再取平均。具体公式如下：
去 周 期 以 后 的 客 流 量 = 客 流 量 / 周 期 因 子 去周期以后的客流量 = 客流量 / 周期因子 去周期以后的客流量=客流量/周期因子
在这里插入图片描述
然后就可以取最后三天的平均， ( 108 + 91.4 + 120 ) / 3 = 106.5 (108+91.4+120)/3=106.5 (108+91.4+120)/3=106.5，作为base。具体取多少天的值，也要通过测试集的表现来确定，也可以按某些函数形式来给每天赋予不同的权重。

其他影响因素

如天气、温度等，针对这些影响因素，可以提取残差，然后用残差训练一个关于这些因素的模型（推荐使用xgboost）。

代码实现

导入库函数：

import pandas as pd 
import sklearn as skr
import numpy as np
import datetime
import matplotlib.pyplot as plt
import seaborn as sns
import warnings 
from dateutil.relativedelta import relativedelta
warnings.filterwarnings('ignore')

数据预处理：

## 加载balance data
def load_data(path: str = 'user_balance_table.csv')->pd.DataFrame:data_balance = pd.read_csv(path)data_balance = add_timestamp(data_balance)return data_balance.reset_index(drop=True)## 添加时间戳
def add_timestamp(data: pd.DataFrame, time_index: str = 'report_date')->pd.DataFrame:data_balance = data.copy()data_balance['date'] = pd.to_datetime(data_balance[time_index], format= "%Y%m%d")data_balance['day'] = data_balance['date'].dt.daydata_balance['month'] = data_balance['date'].dt.monthdata_balance['year'] = data_balance['date'].dt.yeardata_balance['week'] = data_balance['date'].dt.weekdata_balance['weekday'] = data_balance['date'].dt.weekdayreturn data_balance.reset_index(drop=True)## 计算每天的申购总额及赎回总额(2014-03-31之后的)
def get_total_balance(data: pd.DataFrame, date: str = '2014-03-31')->pd.DataFrame:df_tmp = data.copy()df_tmp = df_tmp.groupby(['date'])['total_purchase_amt','total_redeem_amt'].sum()df_tmp.reset_index(inplace=True)return df_tmp[(df_tmp['date']>= date)].reset_index(drop=True)## 生成测试数据
def generate_test_data(data: pd.DataFrame)->pd.DataFrame:total_balance = data.copy()start = datetime.datetime(2014,9,1)testdata = []while start != datetime.datetime(2014,10,15):temp = [start, np.nan, np.nan]testdata.append(temp)start += datetime.timedelta(days = 1)testdata = pd.DataFrame(testdata)testdata.columns = total_balance.columnstotal_balance = pd.concat([total_balance, testdata], axis = 0)total_balance = total_balance.reset_index(drop=True)return total_balance.reset_index(drop=True)## 读取用户信息
def load_user_information(path: str = 'user_profile_table.csv')->pd.DataFrame:return pd.read_csv(path)

## 载入数据 
balance_data = load_data('Dataset/user_balance_table.csv')
balance_data = add_timestamp(balance_data)
total_balance = get_total_balance(balance_data, date = '2014-03-01')
total_balance = generate_test_data(total_balance)
total_balance = add_timestamp(total_balance, 'date')## 创建数据的深层拷贝
data = total_balance.copy() # 2014-03-01到2014-10-14

生成时间序列规则：

## 定义生成时间序列规则预测结果的方法
def generate_base(df: pd.DataFrame, month_index: int)->pd.DataFrame:  # 根据2014-03-01到month_index时间段生成base# 选中固定时间段的数据集total_balance = df.copy()total_balance = total_balance[['date','total_purchase_amt','total_redeem_amt']]s1 = '2014-'+str(month_index)+'-01'total_balance = total_balance[(total_balance['date'] >= '2014-03-01') & (total_balance['date'] < s1)]# total_balance是2014-03-01到2014-month_index-01的申购赎回总额# 加入时间戳total_balance['weekday'] = total_balance['date'].dt.weekday # 星期几total_balance['day'] = total_balance['date'].dt.day # 日total_balance['week'] = total_balance['date'].dt.week # 第几周total_balance['month'] = total_balance['date'].dt.month # 月# 统计周期因子 (获得每日（工作日或周末）均值，再除以整体均值)mean_of_each_weekday = total_balance[['weekday']+['total_purchase_amt','total_redeem_amt']].groupby('weekday',as_index=False).mean()# 计算星期一到星期日的申购赎回总额的平均值for name in ['total_purchase_amt','total_redeem_amt']:mean_of_each_weekday = mean_of_each_weekday.rename(columns={name: name+'_weekdaymean'}) # 重命名列名mean_of_each_weekday['total_purchase_amt_weekdaymean'] /= np.mean(total_balance['total_purchase_amt'])mean_of_each_weekday['total_redeem_amt_weekdaymean'] /= np.mean(total_balance['total_redeem_amt']) # 周一到周日的均值/所有天的均值# 合并统计结果到原数据集total_balance = pd.merge(total_balance, mean_of_each_weekday, on='weekday', how='left') # 左连接# 分别统计周一到周日在(1~31)号出现的频次weekday_count = total_balance[['day','weekday','date']].groupby(['day','weekday'],as_index=False).count()weekday_count = pd.merge(weekday_count, mean_of_each_weekday, on='weekday')# 依据频次对周期因子total_purchase/redeem_amt_weekdaymean进行加权，获得日期因子# 日期因子 = 周期因子*（周一到周日在(1~31)号出现的次数/共有几个月）weekday_count['total_purchase_amt_weekdaymean'] *= weekday_count['date']   / len(np.unique(total_balance['month']))weekday_count['total_redeem_amt_weekdaymean'] *= weekday_count['date']  / len(np.unique(total_balance['month']))day_rate = weekday_count.drop(['weekday','date'],axis=1).groupby('day',as_index=False).sum()# 将训练集中所有日期的均值剔除日期残差得到base# 1~31号的申购赎回总额/日期因子day_mean = total_balance[['day'] + ['total_purchase_amt','total_redeem_amt']].groupby('day',as_index=False).mean()day_pre = pd.merge(day_mean, day_rate, on='day', how='left')day_pre['total_purchase_amt'] /= day_pre['total_purchase_amt_weekdaymean']day_pre['total_redeem_amt'] /= day_pre['total_redeem_amt_weekdaymean'] # 生成测试集数据for index, row in day_pre.iterrows():if month_index in (2,4,6,9) and row['day'] == 31: # 小月没有31号breakday_pre.loc[index, 'date'] = datetime.datetime(2014, month_index, int(row['day'])) # 添加日期列# 基于base与周期因子获得最后的预测结果day_pre['weekday'] = day_pre.date.dt.weekdayday_pre = day_pre[['date','weekday']+['total_purchase_amt','total_redeem_amt']]day_pre = pd.merge(day_pre, mean_of_each_weekday,on='weekday')day_pre['total_purchase_amt'] *= day_pre['total_purchase_amt_weekdaymean'] # baee*周期因子day_pre['total_redeem_amt'] *= day_pre['total_redeem_amt_weekdaymean']day_pre = day_pre.sort_values('date')[['date']+['total_purchase_amt','total_redeem_amt']]return day_pre

生成预测结果：

## 生成预测结果（以及残差）base_list = []
for i in range(4, 10):base_list.append(generate_base(data, i).reset_index(drop=True))base = pd.concat(base_list).reset_index(drop=True)
for i in ['total_purchase_amt','total_redeem_amt']:base = base.rename(columns={i: i+'_base'})data = pd.merge(data.reset_index(drop=True), base.reset_index(drop=True), on='date', how='left').reset_index(drop=True)data['purchase_residual'] = data['total_purchase_amt'] / data['total_purchase_amt_base']data['redeem_residual'] = data['total_redeem_amt'] / data['total_redeem_amt_base'] # 真实值/预测值

结果可视化：

画出预测申购总额和真实申购总额的图像：

## 画出预测申购总额和真实申购总额的图像
real = total_balance[('2014-04-01' <= total_balance['date']) & (total_balance['date'] <= '2014-10-14' )][['date','total_purchase_amt','total_redeem_amt']].reset_index(drop=True)
fig = plt.figure(figsize=(20,6))
plt.plot(base['date'],base['total_purchase_amt_base'] )
plt.plot(real['date'],real['total_purchase_amt'])  
plt.legend(['predict_total_purchase_amt','total_purchase_amt']) 
plt.xlabel("Time")
plt.ylabel("Amount")
plt.show()

在这里插入图片描述
画出预测赎回总额和真实赎回总额的图像：

## 画出预测赎回总额和真实赎回总额的图像
fig = plt.figure(figsize=(20,6))
plt.plot(base['date'],base['total_redeem_amt_base'])
plt.plot(real['date'],real['total_redeem_amt'])  
plt.legend(['predict_total_redeem_amt','total_redeem_amt']) 
plt.xlabel("Time")
plt.ylabel("Amount")
plt.show()

在这里插入图片描述
画出残差的图像：

## 画出残差图像
fig = plt.figure(figsize=(20,6))
plt.plot(data['date'],data['purchase_residual'])
plt.plot(data['date'],data['redeem_residual'])
plt.legend(['purchase_residual','redeem_residual']) 
plt.xlabel("Time")
plt.ylabel("residual")
plt.show()

在这里插入图片描述
可以看出，基于周期因子的时间序列预测较好的拟合了申购、赎回总额的波动趋势。

参考文章：
时间序列规则法快速入门

这篇关于资金流入流出预测—时间序列规则的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/638350。

相关文章

C++统计函数执行时间的最佳实践

C++统计函数执行时间的最佳实践

《C++统计函数执行时间的最佳实践》在软件开发过程中,性能分析是优化程序的重要环节,了解函数的执行时间分布对于识别性能瓶颈至关重要,本文将分享一个C++函数执行时间统计工具,希望对大家有所帮助... 目录前言工具特性核心设计1. 数据结构设计2. 单例模式管理器3. RAII自动计时使用方法基本用法高级用法
阅读更多...
C# LiteDB处理时间序列数据的高性能解决方案

C# LiteDB处理时间序列数据的高性能解决方案

《C#LiteDB处理时间序列数据的高性能解决方案》LiteDB作为.NET生态下的轻量级嵌入式NoSQL数据库,一直是时间序列处理的优选方案,本文将为大家大家简单介绍一下LiteDB处理时间序列数... 目录为什么选择LiteDB处理时间序列数据第一章：LiteDB时间序列数据模型设计1.1 核心设计原则
阅读更多...
MySQL按时间维度对亿级数据表进行平滑分表

MySQL按时间维度对亿级数据表进行平滑分表

《MySQL按时间维度对亿级数据表进行平滑分表》本文将以一个真实的4亿数据表分表案例为基础,详细介绍如何在不影响线上业务的情况下,完成按时间维度分表的完整过程,感兴趣的小伙伴可以了解一下... 目录引言一、为什么我们需要分表1.1 单表数据量过大的问题1.2 分表方案选型二、分表前的准备工作2.1 数据评估
阅读更多...
MySQL中DATE_FORMAT时间函数的使用小结

MySQL中DATE_FORMAT时间函数的使用小结

《MySQL中DATE_FORMAT时间函数的使用小结》本文主要介绍了MySQL中DATE_FORMAT时间函数的使用小结,用于格式化日期/时间字段,可提取年月、统计月份数据、精确到天,对大家的学习或... 目录前言DATE_FORMAT时间函数总结前言mysql可以使用DATE_FORMAT获取日期字段
阅读更多...
Linux中的自定义协议+序列反序列化用法

Linux中的自定义协议+序列反序列化用法

《Linux中的自定义协议+序列反序列化用法》文章探讨网络程序在应用层的实现,涉及TCP协议的数据传输机制、结构化数据的序列化与反序列化方法,以及通过JSON和自定义协议构建网络计算器的思路,强调分层... 目录一，再次理解协议二，序列化和反序列化三，实现网络计算器3.1 日志文件3.2Socket.hpp
阅读更多...
Python标准库datetime模块日期和时间数据类型解读

Python标准库datetime模块日期和时间数据类型解读

《Python标准库datetime模块日期和时间数据类型解读》文章介绍Python中datetime模块的date、time、datetime类,用于处理日期、时间及日期时间结合体,通过属性获取时间... 目录Datetime常用类日期date类型使用时间 time 类型使用日期和时间的结合体–日期时间(
阅读更多...
Java获取当前时间String类型和Date类型方式

Java获取当前时间String类型和Date类型方式

《Java获取当前时间String类型和Date类型方式》：本文主要介绍Java获取当前时间String类型和Date类型方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,... 目录Java获取当前时间String和Date类型String类型和Date类型输出结果总结Java获取
阅读更多...
Python实现批量提取BLF文件时间戳

Python实现批量提取BLF文件时间戳

《Python实现批量提取BLF文件时间戳》BLF（BinaryLoggingFormat）作为Vector公司推出的CAN总线数据记录格式,被广泛用于存储车辆通信数据,本文将使用Python轻松提取... 目录一、为什么需要批量处理 BLF 文件二、核心代码解析：从文件遍历到数据导出1. 环境准备与依赖库
阅读更多...
Spring的RedisTemplate的json反序列泛型丢失问题解决

Spring的RedisTemplate的json反序列泛型丢失问题解决

《Spring的RedisTemplate的json反序列泛型丢失问题解决》本文主要介绍了SpringRedisTemplate中使用JSON序列化时泛型信息丢失的问题及其提出三种解决方案,可以根据性... 目录背景解决方案方案一方案二方案三总结背景在使用RedisTemplate操作redis时我们针对
阅读更多...
go中的时间处理过程

go中的时间处理过程

《go中的时间处理过程》：本文主要介绍go中的时间处理过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1 获取当前时间2 获取当前时间戳3 获取当前时间的字符串格式4 相互转化4.1 时间戳转时间字符串 (int64 > string)4.2 时间字符串转时间
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2