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

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。

相关文章

Java controller接口出入参时间序列化转换操作方法(两种)

Java controller接口出入参时间序列化转换操作方法(两种)

《Javacontroller接口出入参时间序列化转换操作方法(两种)》：本文主要介绍Javacontroller接口出入参时间序列化转换操作方法,本文给大家列举两种简单方法,感兴趣的朋友一起看... 目录方式一、使用注解方式二、统一配置场景：在controller编写的接口，在前后端交互过程中一般都会涉及
阅读更多...
python处理带有时区的日期和时间数据

python处理带有时区的日期和时间数据

《python处理带有时区的日期和时间数据》这篇文章主要为大家详细介绍了如何在Python中使用pytz库处理时区信息,包括获取当前UTC时间,转换为特定时区等,有需要的小伙伴可以参考一下... 目录时区基本信息python datetime使用timezonepandas处理时区数据知识延展时区基本信息
阅读更多...
Python的time模块一些常用功能(各种与时间相关的函数)

Python的time模块一些常用功能(各种与时间相关的函数)

《Python的time模块一些常用功能(各种与时间相关的函数)》Python的time模块提供了各种与时间相关的函数,包括获取当前时间、处理时间间隔、执行时间测量等,：本文主要介绍Python的... 目录1. 获取当前时间2. 时间格式化3. 延时执行4. 时间戳运算5. 计算代码执行时间6. 转换为指
阅读更多...
Java中字符串转时间与时间转字符串的操作详解

Java中字符串转时间与时间转字符串的操作详解

《Java中字符串转时间与时间转字符串的操作详解》Java的java.time包提供了强大的日期和时间处理功能,通过DateTimeFormatter可以轻松地在日期时间对象和字符串之间进行转换,下面... 目录一、字符串转时间（一）使用预定义格式（二）自定义格式二、时间转字符串（一）使用预定义格式（二）自
阅读更多...
Nginx location匹配模式与规则详解

Nginx location匹配模式与规则详解

《Nginxlocation匹配模式与规则详解》：本文主要介绍Nginxlocation匹配模式与规则,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、环境二、匹配模式1. 精准模式2. 前缀模式（不继续匹配正则）3. 前缀模式（继续匹配正则）4. 正则模式（大
阅读更多...
Java中Date、LocalDate、LocalDateTime、LocalTime、时间戳之间的相互转换代码

Java中Date、LocalDate、LocalDateTime、LocalTime、时间戳之间的相互转换代码

《Java中Date、LocalDate、LocalDateTime、LocalTime、时间戳之间的相互转换代码》：本文主要介绍Java中日期时间转换的多种方法,包括将Date转换为LocalD... 目录一、Date转LocalDateTime二、Date转LocalDate三、LocalDateTim
阅读更多...
golang获取当前时间、时间戳和时间字符串及它们之间的相互转换方法

golang获取当前时间、时间戳和时间字符串及它们之间的相互转换方法

《golang获取当前时间、时间戳和时间字符串及它们之间的相互转换方法》：本文主要介绍golang获取当前时间、时间戳和时间字符串及它们之间的相互转换,本文通过实例代码给大家介绍的非常详细,感兴趣... 目录1、获取当前时间2、获取当前时间戳3、获取当前时间的字符串格式4、它们之间的相互转化上篇文章给大家介
阅读更多...
Feign Client超时时间设置不生效的解决方法

Feign Client超时时间设置不生效的解决方法

《FeignClient超时时间设置不生效的解决方法》这篇文章主要为大家详细介绍了FeignClient超时时间设置不生效的原因与解决方法,具有一定的的参考价值,希望对大家有一定的帮助... 在使用Feign Client时，可以通过两种方式来设置超时时间：1.针对整个Feign Client设置超时时间
阅读更多...
springboot+dubbo实现时间轮算法

springboot+dubbo实现时间轮算法

《springboot+dubbo实现时间轮算法》时间轮是一种高效利用线程资源进行批量化调度的算法,本文主要介绍了springboot+dubbo实现时间轮算法,文中通过示例代码介绍的非常详细,对大家... 目录前言一、参数说明二、具体实现1、HashedwheelTimer2、createWheel3、n
阅读更多...
详解nginx 中location和 proxy_pass的匹配规则

详解nginx 中location和 proxy_pass的匹配规则

《详解nginx中location和proxy_pass的匹配规则》location是Nginx中用来匹配客户端请求URI的指令,决定如何处理特定路径的请求,它定义了请求的路由规则,后续的配置（如... 目录location 的作用语法示例：location /www.chinasem.cntestproxy
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2