在数据库里试试这个笔记本介绍金融机构的数据科学家、数据工程师和分析师的作用包括（但不限于）保护价值数千亿美元的资产，保护投资者免受万亿美元的冲击，比如说从闪电崩盘中。这些问题背后最大的技术挑战之一是时间序列的缩放操作。Tick数据、地理空间或交易数据等替代数据集以及基本经济数据都是金融机构可用的丰富数据源的例子，所有这些数据都是按时间戳自然索引的。解决金融领域的业务问题（如风险、欺诈和合规性）最终取决于能够并行地聚合和分析数千个时间序列。基于关系数据库的旧技术在分析交易策略或对多年的历史数据进行监管分析时不容易扩展。此外，许多现有的时间序列技术使用专门的语言，而不是基于标准SQL或Python的api。幸运的是，Apache Spark™ 它包含大量的内置功能，例如自然地并行时间序列操作的窗口。此外，Koalas是一个开源项目，允许您使用熟悉的pandas语法通过Apache Spark执行分布式机器学习查询，有助于将这种能力扩展到数据科学家和分析师。在这个博客中，我们将展示如何在数十万个股票代码上并行构建时间序列函数。接下来，我们将演示如何在本地IDE中模块化函数，并使用Databricks Connect创建丰富的时间序列特性集。最后，如果您是一个pandas用户，希望通过缩放数据准备来进行财务异常检测或其他统计分析，我们将使用一个市场操纵示例来说明考拉如何使缩放对典型的数据科学工作流透明。设置时间序列数据源让我们从吸收两个传统的金融时间序列数据集开始：交易和报价。我们模拟了这个博客的数据集，这些数据集是根据从交易报告工具（trades）和国家最佳出价（NBBO）feed（来自纽约证券交易所等交易所）的数据为模型。您可以在这里找到一些示例数据：https://www.tickdata.com/product/nbbo/。本文通常假设基本财务条款；有关更广泛的参考资料，请参阅Investopedia的文档。从下面的数据集中值得注意的是，我们已经为每个时间戳分配了timestamp类型，因此出于规范化的目的，交易执行时间和报价更改时间被重命名为event_ts。此外，如本文所附的完整笔记本所示，我们最终将这些数据集转换为Delta格式，以便确保数据质量并保持列格式，这对于下面的交互式查询类型最为有效。贸易模式=结构类型([StructField（"symbol"，StringType（）），StructField（"事件"，TimestampType（）），StructField（"trade_dt"，StringType（）），StructField（"trade_pr"，DoubleType（））])quote_schema=结构类型([StructField（"symbol"，StringType（）），StructField（"事件"，TimestampType（）），StructField（"trade_dt"，StringType（）），StructField（"bid_pr"，DoubleType（）），StructField（"ask_pr"，DoubleType（））])用apachespark合并和聚合时间序列™今天全球金融市场上有超过60万种公开交易的证券。鉴于我们的交易和报价数据集涵盖了这么多证券，我们需要一个易于扩展的工具™ 为ETL提供了一个简单的API，它是并行化的标准引擎，它是我们合并和聚合标准度量的工具，反过来帮助我们了解流动性、风险和欺诈。我们将从合并交易和报价开始，然后聚合交易数据集以显示分割数据的简单方法。最后，我们将展示如何使用Databricks Connect将这些代码打包到类中，以便更快地进行迭代开发。用于以下指标的完整代码在随附的笔记本中。连接状态as-of-join是一种常用的"merge"技术，它返回在左时间戳时有效的最新right值。对于大多数时间序列分析，多种类型的时间序列在符号上连接在一起，以了解一个时间序列（例如NBBO）在另一个时间序列（例如交易）中的特定时间的状态。以下示例记录了所有符号的每次交易的NBBO状态。如下图所示，我们从初始基准时间序列（交易）开始，合并了NBBO数据集，以便每个时间戳都有"截至交易时"记录的最新买入价和卖出价。一旦我们知道了最新的买入和卖出价，我们就可以计算差额（称为价差），以了解流动性在什么时候可能较低（以较大的价差表示）。这种指标会影响你如何组织你的交易策略来提高你的α值。首先，让我们使用内置的窗口函数last来查找按时间排序后的最后一个非空引号值。#联接方法内的示例代码#为窗口定义分区键分区规格=窗口.分区依据（"符号"）#define sort-ind_cd是一个排序键（交易前报价）join_spec=分区_规范订购人（"事件"）。\划船(Window.unboundedPreceding窗口, 窗口.currentRow)#使用"最后一个值"功能获取最新的有效记录select（last（"bid"，True）.over（join_spec）.alias（"latest_bid"））现在，我们将调用我们的自定义联接来合并我们的数据并附加我们的报价。完整代码见随附笔记本。#应用我们的自定义联接市场交易=贸易过滤器（col（"symbol"）=="K"）mkt_hrs_trades_ts=基本交易（市场交易）引号=引号.filter（col（"symbol"）=="K"）显示（市场交易_ts.加入（引用） 根据贸易模式标记VWAP我们已经展示了上面的一种合并技术，现在让我们关注一个标准的聚合，即体积加权平均价格（VWAP），即按体积加权的平均价格。这个指标是一个指标的趋势和价值的证券全天。vwap函数在我们的包装类（在所附的笔记本）显示vwap下降到高于或低于证券的交易价格。特别是，我们现在可以确定VWAP（橙色）跌破交易价格的窗口，表明该股票已超买。贸易=基本(交易.select（"事件"，符号，"价格"，lit（100）.alias（"volume"））vwap_df=交易_ts.vwap公司（频率='m'）显示器（vwap_测向滤波器（col（符号）="K"）\.filter（col（'time_group'）.between（'09:30'，'16:00'））\.orderBy（'time_group'））使用Databricks Connect加快迭代开发到目前为止，我们已经为一次性时间序列度量创建了一些基本的包装器。然而，代码的产品化需要模块化和测试，这最好在IDE中实现。今年，我们推出了Databricks Connect，它提供了本地IDE开发的能力，并增强了针对实时Databricks集群进行测试的经验。Databricks Connect for financial Analysis的好处包括能够在小型测试数据上添加时间序列特征，并增加了针对多年历史记录执行交互式Spark查询的灵活性用于验证功能的数据。我们使用PyCharm来组织包装PySpark功能所需的类，以生成丰富的时间序列特性集。这个IDE为我们提供了代码完成、格式化标准和一个在运行代码之前快速测试类和方法的环境。我们可以快速调试类，然后使用Jupyter笔记本直接从我们的笔记本上运行Spark代码，它加载我们的本地类，并使用可伸缩的基础设施执行交互式查询。控制台窗格显示针对活动集群执行的作业。最后，通过使用我们的本地IDE，同时在我们最大的时间序列数据集上附加到我们的物化时间序列视图中，我们得到了两者的最佳效果。利用考拉操纵市场pandas API是Python中用于数据操作和分析的标准工具，并深入集成到Python数据科学生态系统中，例如NumPy、SciPy、matplotlib。pandas的一个缺点是它不容易扩展到大量数据，财务数据总是包含多年的历史数据，这对于风险汇总或合规性分析至关重要。为了简化这一点，我们引入了考拉作为一种在后端执行Spark的同时利用pandas api的方法。由于考拉API与Pandas相匹配，所以我们不会牺牲易用性，而迁移到可伸缩代码只需一行代码更改（请参阅下一节中的考拉导入）。在我们展示考拉适合金融时间序列问题之前，让我们先从金融欺诈中的一个特定问题的背景开始：抢先。发生以下顺序时，发生前运行：交易公司知道可能影响证券价格的非公开信息该公司购买了大量的订单（或一系列订单，总的数量很大）由于流动性的消除，证券价格上涨该公司将证券出售给投资者（从上一次的购买中一直向上推高），并赚取巨额利润，迫使投资者支付更高的价格，即使证券交易所依据的信息是非公开的来源：CC0公共域图片：https://pxhere.com/en/photo/1531985, https://pxhere.com/en/photo/847099为了便于说明，这里有一个简单的例子，使用农贸市场和苹果派生意。这个例子展示了弗雷迪，一个跑步者，他意识到全国苹果派企业对苹果的需求迫在眉睫，随后在所有农贸市场购买苹果。事实上，这使得弗雷迪能够以溢价的价格向买家出售他的苹果，因为弗雷迪先于任何其他买家购买而造成了重大影响（抑制）