北風網項目實戰培訓第十三講、金融中的大數據應用及Python實現講師：朱彤（北風網版權所有)Python金融應用概述隨著云計算技術的出現，大數據應用開始逐步有所發展。那么什么是大數據，你又應當如何通過大數據來獲得有用的信息？大數據與我們每天遇到的通常的數據形式有什么不同？本章將主要回答這些問題，并且向你介紹金融中大數據的使用。大數據工具提供了分析大量數據的可靠性工具，這些數據可能來自于多個來源。為了實現大數據需求，ApacheHadoop成為了金融機構及公司的最主要選擇。這樣，對于金融工程師而言，了解Hadoop用于金融分析的方式是非常關鍵的。在我們開始處理大數據集的時候，我們還需要找到一種方式來存儲數據。最常規的存儲數據的方式是結構化查詢語言（SQL）這樣的關系型數據庫管理系統。電子數據的本質是總會發生變化的，而其他的一些存儲數據的方式就被引入，特別是關于非SQL的產品。非關系型數據庫機制的一種典型應用是NoSQL，稱為NotOnlySQL。除了使用SQL類似的語言進行數據管理意外，NoSQL允許存儲非結構化的數據，例如鍵值對，圖形或者文件。因為從設計角度的簡單，它也可以在某些環境下表現的更為有效。金融中NoSQL的一個應用方式是存儲tick數據。本章將介紹使用NoSQL存儲tick數據的方式。概述本章我們將介紹如下話題：介紹大數據，ApacheHadoop以及其組成成分獲得Hadoop并且運行一個快速啟動的虛擬機使用Hadoop的HDFS文件存儲使用Python和MapReduce結合實現一個對電子書的單詞計數的程序。在Hadoop運行之前對MapReduce程序的檢驗對于某只股票日度的價格變化進行MapReduce操作使用Python分析MapReduce操作的結果介紹NoSQL獲取并運行MongoDB獲取和安裝Python使用的PyMongo介紹PyMongo的數據庫和集合使用NoSQL集合進行tick數據的插入，刪除，查找和存儲操作。了解大數據關于大數據有很多激動人性的特性包含在其中。在開始本章的介紹之前，非常重要的一個工作室定義大數據以及了解你使用大數據來獲得有用信息的方式。大數據與我們日常碰到的數據有什么不同，例如新聞故事，報告，文獻以及聲音等相比有什么不同？大數據實際上用高速來刻畫的，它會累計到一個較大的數據量，通常是TB或PB級別。常規的軟件工具已經無法在短時間內獲取，處理，操作和管理這樣的數據。應用于這些大規模的數據集的分析性的工具是用來揭示數據的信息及其之間的關系，以更好的用于預測和其他分析性的行為。隨著云計算的出現，大數據變得越來越平常。大量的信息可以以較低的成本存儲在云端。這推動了從關系型的存儲到非關系型的解決方案的過渡，例如NoSQL，允許非關系型的數據以較快的速率被獲取。通過在獲取的數據上進行數據分析，公司可以改進運行的效率，分析的模式，運行目標的市場營銷計劃并改進客戶的感知。金融行業的公司正在逐步將大數據整合到他們的運行當中，例如，實時分析客戶的交易來確定異常的行為和發現欺詐。客戶的記錄，消費的習慣甚至是社交媒體的行為可以用來通過市場分割來改進產品和服務。大數據同居提供了分析風險和信用方面大數據的可靠框架，而這樣的數據又是來源于很多渠道。Hadoop用于大數據分析ApacheHadoop是一個100%開源的框架，其主要功能有兩個：存儲和處理大數據。它逐步成為對多個服務器中分布式并行處理數據的領先大數據工具，而且可以無限的擴大規模。因為規模擴展性，靈活性，容錯性以及低成本的特性，許多云計算解決方案提供商，金融機構以及公司都使用Hadoop來滿足其大數據需求。Hadoop框架包含完成其功能的關鍵模塊：Hadoop分布式文件系統（HDFS），YetAnotherResourceNegotiator（YARN）以及MapReduce（MapR）。Hadoop用于大數據分析HDFS：HDFS是Hadoop特有的文件系統，設計用來規模擴展和可移植操作，允許大量的數據存儲在Hadoop集群的多個節點上，數據可以是GB規模甚至是TB規模。集群中的數據被劃分為小的128Mb的小的模塊，并分布在各個集群上。MapReduce數據處理函數用來對大數據集合的小的子集進行操作，因此提供了大數據處理所需要的規模擴展操作。HDFS文件系統使用TCP/IPsocket通信來作為一個大的文件系統在網絡中提供數據。YARN：YARN是一個資源管理和計劃平臺管理CPU，內存以及在Hadoop集群中運行的應用的存儲。它包含負責如下功能的模塊：在相同集群運行的應用之間分配資源同時滿足一定約束，例如隊列功能和用戶限制等的功能；基于每個應用的資源需求計劃任務；協調計劃的合適資源；跟蹤和監控應用程序的運行以及其資源使用。MapReduce：MapReduce是Hadoop用于處理和生成大數據集合的軟件程序框架。對于開發者來說，MapReduce可能是Hadoop中最重要的程序設計組成。它包含兩個功能，map和reduce。Map功能處理鍵值對來產生中間鍵值對，reduce功能將所有的具有相同中間鍵的中間鍵值對合并起來來得到結果。MapReduce減輕了軟件需要處理的跨網絡移動數據的需求，讓處理軟件直接面對數據。MapReduce主要是使用Java進行處理，其他的語言，例如SQL和Python，也可以對MapReduce進行實現，使用Hadoop流工具。大數據適合我么？也許你正在運行一個文件，大約幾百兆的規模，而你目前的數據分析工具運行的過于緩慢。因為大數據通常包含TB或GB級別的存儲，那么你可能會問，ApacheHadoop在這種情況下適合我么？注意Hadoop實現的常規計算時將每一個你數據中的輸入進行映射，然后進行某些簡化計算的操作來加總個體的部分。你事實上可以使用編程語言，例如SQL和Python來實現相同的分組和計數工作。另外，存儲節點允許你更為容易的表達計算流。或者你也許希望考慮移植數據分析工具到pandas或R，它們也是非常強大的，可以處理GB級別的數據，編碼是有效的，且沒有內存泄漏。許多商業的SQL服務器也可以完成這項任務。除此之外，內存和存儲成本是可以接受的，這樣你可以在自己的工作站上運行大規模的計算。如果你的數據是TB級別的，那么就沒有那么幸運，沒有其他的一些選擇，那么ApacheHadoop以及其他的大數據處理工具例如ApacheSpark看似就是最好的選擇，它們會提供規模可擴展性，可支付以及容錯性。獲取ApacheHadoopApacheHadoop的官方網頁是，這里，你可以找到深入的文檔，手冊以及ApacheHadoop的發布版本。Hadoop是用Java寫的，需要安裝JVM到你的單一節點運行，它支持GNU/Linux以及Windows。因為本章的主要目的是介紹與ApacheHadoop相關的Python編程，理想的是有一種快速的獲取完整的Hadoop生態系統的方式。云計算提供商Cloudera提供一組免費的QuickStartVM，包含單一節點的ApacheHadoop集群，完整的樣本示例以及幫助我們直接管理集群的工具。因此，我們在這里使用虛擬機的方式。下載地址/content/support/en/downloads/quickstart_vms.html因為VM是64位的，因此需要64位的基礎操作系統以及支持64位的虛擬機，需要4GB的內存。運行的虛擬機的用戶名和密碼都是cloudera，這個包含root權限，主目錄是/home/clouderaHadoop處理一個單詞計數程序最為簡單的一種了解Hadoop編程的方式是一個簡單的對相對比較大的電子書的單詞計數功能。Map程序會讀入文本中的每一行，以空格或制表符分割并返回一個鍵值對，默認的計數為1。而程序reduce會讀入來自于map程序的說有鍵值對，并且對相似的單詞進行加總。Hadoop的產出是一個輸出文件，包含整個電子書單詞的一個列表以及單詞出現的次數。下載樣本數據：Gutenberg提供了超過100000本免費的電子書，以HTML，EPUB，Kindle以及純文本UTF-8的格式存儲。我們在這里使用電子書進行測試，我們選用由JamesJoyce所編著的Ulysses一書。純文本UTF-8文件的路徑是http://www.G/ebooks/4300.txt.utf-8，使用Firefox或者其他的網絡瀏覽器，我們可以下載這個文件，并將其保存在Downloads文件夾中，文件名為pg4300.txt。在將文件下載到磁盤以后，我們將它復制到HadoopHDFS文件存儲中進行處理。這樣的操作會將電子書保存到Hadoop的HDFS存儲，名稱相同為pg4300.txt。為了保證拷貝操作的成功，我們使用Hadoopfs–ls命令來看一下文件。這樣可以看到Hadoop文件存儲成功的保存了我們的文件。Hadoop處理一個單詞計數程序我們下面要在Hadoop上運行Python編寫的map程序，編寫一個名為mapper.py的文件，可以在后面使用。當運行mapper.py的時候，解釋器會讀入輸入緩沖區的來自于文本的數據。所有的文本都被使用空白字符進行分隔，每個單詞被賦值為計數一次，使用制表符進行分隔輸出。CentOS中的一個用戶友好的編輯器是gedit，創建一個名為word_count的目錄，保存文件為/home/cloudera/word_count/mapper.py。程序文件mapper.py需要被指定為Linux系統下可執行的文件，從終端使用一個命令可以進行此項操作。下面我們創建reduce.py程序，也存放在/home/cloudera/word_count/目錄下。這里，reduce程序從map程序讀入所有的鍵值對。當前的單詞以及單詞的技術通過分隔符制表符來獲得，并且和之前出現的單詞對比。如果相同的單詞，則增加單詞的計數加1，然后返回結果，打印單詞本身以及計數的數量，用制表符分隔。再一次的，應當把reduce.py也指定為是Linux可運行的程序。這樣可以保證我們的程序沒有限制的運行。Hadoop處理一個單詞計數程序在Hadoop中運行我們的map和reduce程序之前，我們可以在本機上首先運行，保證程序正常的工作。我們可以使用圖形界面的方式看到我們剛剛建立的兩個python文件。在終端，我們首先運行一下mapper.py這個Python文件。和預期的一樣，map程序給出了每個讀入的單詞，設定計數為1，并且打印了每一行的結果。在終端，運行reduce.py程序。我們看到，reduce程序對于map程序中出現的相似單詞的計數進行了相加，并打印了每行的結果。Hadoop處理一個單詞計數程序在Hadoop中運行MapReduce：我們現在可以在Hadoop上運行我們的MapReduce操作，在終端使用命令運行。這里，我們使用Hadoop的Streaming工具來應用我們的Python代碼作map和reduce的操作，在這里進行HadoopStreaming操作的JavaJAR文件為hadoop-streaming-2.6.0-mr1-cdh5.5.0.jar，存放在/usr/lib/hadoop-0.20-mapreduce/contrib/streaming/目錄下。其他的系統中，HadoopStreaming的JAR文件可能存放在其他目錄，或者使用不同的文件名。通過提供其他的參數，指定了輸入文件，輸出的目錄，這些都是Hadoop操作所必須的。在運行Hadoop時，終端中的輸出告訴了我們一些有趣的事實，包括MapReduce操作可以在http://quickstart.cloudera:8088/proxy/application_1417846146061_0002進行跟蹤，也告訴我們操作的進程，目前50%，因為整個運行需要一定的時間。在結束的時候，會出現文件的大小，這告訴我們成功的運行了程序，而結果被寫入到輸出文件夾pg4300-output中，我們可以看一下HDFS文件系統來找到這個輸出。我們看到在文件存儲中加入了一個新的文件夾，這是我們的輸出文件夾，我們可以看一下這個文件夾的內容。Hadoop處理一個單詞計數程序這里，我們可以看到目標文件夾中Hadoop給出了兩個文件，_SUCCESS文件是空文件，告訴我們Hadoop所進行的MapReduce操作成功的完成。第二個文件是part-00000，包含reduce操作的結果，我們可以使用fs–cat命令來查看。使用Hue來瀏覽Hadoop：除了使用終端瀏覽HDFS文件系統以外，另外一種方式使用GUI接口來查看HDFS文件系統，這可以通過Hueweb接口管理器來進行。Hue也是觀察大量輸出信息的最好方式，例如觀察所有的單詞計數操作的結果。地址是：http://quickstart.cloudera:8888/filebrowser/深入學習——金融中的Hadoop現在我們已經知道了如何使用Hadoop進行簡單的對一個較大的文件進行單詞計數工作。接下來我們深入一步，使用Hadoop進行量化金融分析。首先我們來計算某股票價格變化的日內歷史百分比。從Yahoo財經獲取IBM股票價格數據：為了獲得數據集，我們可以使用Yahoo金融提供的歷史股票價格數據。使用Firefox或其他CentOS環境中的web瀏覽器，我們可以下載日度的價格數據，存儲為CSV文件，地址是:/table.csv?a=IBM在這個例子中，我們使用IBM的股票數據，將這個文件下載到用戶主目錄的Downloads文件夾下，并重命名為ibm.csv，如果我們看一下這個文件，會發現日度的價格是今天直到1962年。接下來將目標CSV文件拷貝到Hadoop的HDFS文件系統。如果在HDFS文件列表中，有這個文件，則表明操作成功。深入學習——金融中的Hadoop接下來要做的事情是創建我們的map和reduce程序來研究股票價格。在我們的主目錄，創建一個新的目錄，稱為stock，我們可以將前一節所寫的mapper.py和reduce.py拷貝到文件夾/home/cloudera/stock/下。使用gedit或其他文本編輯器打開mapper.py來編輯我們的程序。修改程序告訴我們的map程序在讀取CSV文件的時候忽略第一行，因為第一行包含冗余的頭信息。下面的行包含日期，開盤價，最高價，最低價，收盤價，交易量和調整的收盤價，使用逗號分隔。我們關心的是使用開盤價和收盤價計算的一日價格的變化。程序輸出一位小數表示的值，使用逗號隔開。我們將使用相同的reduce程序，不進行任何改變。保證兩個文件都可以被執行，這需要在終端中使用命令進行權限改變。在進行Hadoop處理之前，我們先在系統中對程序進行運行，我們檢驗mapper.py程序。這會給我們一個很長的價格變化的列表。這樣我們看到map程序的結果非常類似于之前的單詞計數程序。深入學習——金融中的Hadoop現在我們就可以在Hadoop中運行map和reduce程序了，這里使用的Unix命令類似于之前的單詞計數程序，只有很少的改變。這樣Hadoop的MapReduce操作就開始運行。當這個運行結束以后，我們可以看到簡略的結果。和預期一樣，輸出結果會被存放到HDFS文件系統的stock-output目錄下。在進行結果的研究分析之前，我們需要將這個HDFS文件系統中的文件拷貝到本地的工作目錄，在終端可以使用命令進行。這樣我們可以使用這個文件。使用Hue，我們還是可以觀察輸出的結果。在使用Python對結果進行繪圖之前，我們需要首先安裝matplotlib模塊，這很簡單，在終端使用安裝命令就可以。這需要一定的時間，結束以后，matplotlib模塊就安裝到了我們的系統中。我們編寫一個python文件，來繪制結果的柱狀圖。讀入輸出中的每一行，將這個結果分隔為x和y值的列表中，然后使用matplotlib的bar函數進行繪圖。width參數給出了柱子的寬度，這里我們縮小到0.1，以實現完全的表示。我們還關心的是最大和最小的IBM股票的價格變化，這些數值我們進行返回。將這個文件命名為analysis.py，并存儲在/home/cloudera/stock/目錄下。從終端中，我們可以運行這個文件。總體來看，日內IBM的價格變化最大是23.5%，最小是-13%，日內變化的分布以0.0為中心。NoSQL許多云計算提供商都提供類似于NoSQL的存儲工具，在模型的文檔類型中存儲非結構化的數據。本節，我們將探討NoSQL作為一種金融數據存儲方式的實現。有許多開源的數據庫免費的支持NoSQL存儲。MongoDB是一個用C++編寫的開源的文檔數據庫。MongoDB的官方網站是。MongoDB有Linux，Windows，MacOSX以及Solaris的版本。我們需要到/downloads下載和安裝MongoDB到你的本地機器。MongoDB管理服務在安裝完成的時候是默認的增加到你的操作系統的實時環境中，可以使用命令行來運行。在官方文檔中還包括如何開啟mongodb服務以及在你的機器上運行的指南。創建數據目錄和運行MongoDB：在首次運行MongoDB之前，我們需要創建一個目錄，其中mongod過程寫入數據。Mongod過程默認使用/data/db作為默認你目錄。我們在選擇的目錄下創建一個data目錄，保證你的用戶賬戶對此目錄有讀和寫的權限。在Windows中，打開命令行模式，移動到你的工作目錄，創建數據目錄，并啟動mongod服務。在我們的例子中，安裝文件存儲在c:\mongos，工作目錄是c:\test。MongoDB獲得PyMongo：PyMongo模塊包含使用Python與MongoDB進行交互的工具。PyMongo的官方主頁是/pypi/pymongo/。這里包含了安裝指南，我們會看到，安裝的過程非常類似于常規的在Windows，Linux，或MacOSX上安裝Python包的過程。最簡單的安裝PyMongo的方式是下載項目源文件，解壓到本地硬盤，然后使用終端跟蹤到這個目錄，運行pythonsetup.pyinstall來進行安裝。另外，如果你的系統中安裝有pip，也可以使用它來進行安裝。運行連接檢測：我們來進行一個簡單的連接檢測，保證我們的MongoDB服務以及PyMongo模塊正確的安裝。獲取數據庫：使用單一的數據目錄data/db，我們可以創建過個獨立的數據庫。在PyMongo的幫助下，訪問數據庫可以通過一個簡單的MongoDB實例來實現。例如，如果你希望數據庫的名字叫ticks_db，來存儲某些tick數據，那么可以很容易的創建。如果數據庫的命名方式不支持，那么也是可以訪問的。MongoDB獲取一個集合：集合是一組文件，存儲在數據庫中，類似于關系型數據庫的表格。一個文件可以接受不同的輸入類型，使用二元JSON（BSON）格式來設定，包括基本類型例如字符串和整數，Python列表形式表示的數組，UTF-8編碼的Unicode的二元字符串，ObjectId，也就是唯一的標識符以及Python字典對象。記住每個文檔可以包含最多16MB數據。在我們的李子中，我們感興趣的是存儲AAPL股票的tick數據，我們在一個名為aapl的集合中存儲tick數據。類似于訪問數據庫，你對集合的命名如果不被支持，例如，命名為aapl-collection，那么也是可以被用特殊方式訪問的。注意如果對象插入到的數據庫或者集合是不存在的，那么會自動創建。MongoDB插入一個文檔：假設我們現在針對的是AAPL的市場tick數據進行操作，我們感興趣的是如何存儲這些數據。我們將tick數據結構化為一個Python的字典，存儲tick的名稱，收到的時間，開盤，最高，最低和最后的價格，以及總交易量等交易信息。假設我們有一條10AM收到的數據。我們可以簡單的使用insert函數在集合中插入一個字典對象，這樣由insert函數會生成一個唯一的鍵，有一個特殊的域命名為_id。在插入了第一個文檔之后，aapl集合會在服務器上創建，我們可以列出數據庫中所有的集合來證明這一點。獲取單一的文檔：最基本的對集合中文檔的查詢是使用find_one函數，使用這個方法，沒有任何參數，可以給出我們第一個匹配的項目，如果沒有匹配，則是None。否則，參數可以接收一個字典作為過濾的條件。我們可以使用多種方法返回相同的結果，注意的是，ObjectId屬性需要被轉化為字符串來訪問服務器生成的_id字段。刪除文檔：remove函數可以刪除集合中匹配的文檔MongoDB批量插入文檔：我們可以使用insert函數接收一個逗號分隔的字典列表。我們現在加入兩個假想的后兩個時間點的tick價格數據到我們的集合中。集合中文檔的計數：count函數可以用于任何查詢來計算符合條件的個數，也可以與find函數共同使用。查找文檔：find函數類似于find_one函數，除了它是返回一個文檔的列表，沒有其他的參數，find函數簡單的返回所有集合中的項目。同樣，也可以對tick數據的集合進行過濾搜索，例如，我們可以找到10:02之前的兩條數據。在find.sort的例子中，我們排序了搜索結果，按照時間升序的順序，我們也可以進行降序排序。MongoDB總結：使用PyM