New data process api; adding indicators

Author: pritam001

atx-data-processor.postman_collection.json

@@ -239,7 +239,7 @@
 				"header": [],
 				"body": {
 					"mode": "raw",
-					"raw": "{\n\t\"localCsvUrl\": \"~/Downloads/28-02-2020-TO-13-03-2020HDFCALLN.csv\",\n\t\"useWorkers\": true\n}",
+					"raw": "{\n\t\"localCsvUrl\": \"~/Downloads/29-03-2018-TO-27-03-2020HDFCALL.csv\",\n\t\"useWorkers\": true\n}",
 					"options": {
 						"raw": {
 							"language": "json"
@@ -260,6 +260,46 @@
 			},
 			"response": []
 		},
+		{
+			"name": "post nse/process",
+			"event": [
+				{
+					"listen": "prerequest",
+					"script": {
+						"id": "9c74e120-8216-44d4-80d3-b9ef99a6a87a",
+						"exec": [
+							""
+						],
+						"type": "text/javascript"
+					}
+				}
+			],
+			"request": {
+				"method": "POST",
+				"header": [],
+				"body": {
+					"mode": "raw",
+					"raw": "{\n\t\"symbol\": \"HDFC\"\n}",
+					"options": {
+						"raw": {
+							"language": "json"
+						}
+					}
+				},
+				"url": {
+					"raw": "{{atx-data-processor-url}}/{{atx-data-processor-api}}/nse/process",
+					"host": [
+						"{{atx-data-processor-url}}"
+					],
+					"path": [
+						"{{atx-data-processor-api}}",
+						"nse",
+						"process"
+					]
+				}
+			},
+			"response": []
+		},
 		{
 			"name": "job/all/failed/remove",
 			"event": [
@@ -526,13 +566,13 @@
 	],
 	"variable": [
 		{
-			"id": "a2ccee22-174b-4dac-9cb9-e2c9c28a5d76",
+			"id": "b5a8fa65-5a22-44b0-ac95-8b2413146ec3",
 			"key": "atx-data-processor-api",
 			"value": "",
 			"type": "string"
 		},
 		{
-			"id": "d8562c3a-528c-4549-b9e2-4c9218b2dbb8",
+			"id": "be297e39-4431-41d2-bf60-32633fa82fed",
 			"key": "atx-data-processor-url",
 			"value": "",
 			"type": "string"

src/lib/__init__.py

@@ -0,0 +1,53 @@
+"""
+    Indicators as shown by Peter Bakker at:
+    https://www.quantopian.com/posts/technical-analysis-indicators-without-talib-code
+"""
+
+# Import Built-Ins
+import logging
+
+# Import Third-Party
+import pandas as pd
+import numpy as np
+
+# Init Logging Facilities
+log = logging.getLogger(__name__)
+
+
+def moving_average(df, p, n):
+    """Calculate the moving average for the given data.
+
+    :param df: pandas.DataFrame
+    :param p: String
+    :param n: Integer
+    :return: pandas.DataFrame
+    """
+    MA = pd.Series(df[p].rolling(n, min_periods=n).mean(), name="MA_" + str(n))
+    df = df.join(MA)
+    return df
+
+
+def exponential_moving_average(df, p, n):
+    """
+
+    :param df: pandas.DataFrame
+    :param p: String
+    :param n: Integer
+    :return: pandas.DataFrame
+    """
+    EMA = pd.Series(df[p].ewm(span=n, min_periods=n).mean(), name="EMA_" + str(n))
+    df = df.join(EMA)
+    return df
+
+
+def momentum(df, p, n):
+    """
+
+    :param df: pandas.DataFrame
+    :param p: String
+    :param n: Integer
+    :return: pandas.DataFrame
+    """
+    M = pd.Series(df[p].diff(n), name="Momentum_" + str(n))
+    df = df.join(M)
+    return df

src/lib/technical_indicators.py

@@ -0,0 +1,31 @@
+from sqlalchemy import Column, Float, TIMESTAMP, VARCHAR
+from sqlalchemy.ext.declarative import declarative_base
+
+from src.model.nse_model import NseDailyDataModel
+
+Base = declarative_base()
+
+
+class NseDailyDataProcessedModel(NseDailyDataModel):
+    """Model for Processed NSE Daily Data"""
+
+    __tablename__ = "nse_data_daily_processed"
+
+    ma = Column(Float)
+
+    def __repr__(self):
+        return (
+            "<NseDailyDataProcessedModel(timestamp='%s', open='%s', close='%s', high='%s', low='%s', volume='%s',"
+            " symbol='%s', createdTime='%s', updatedTime='%s')>"
+            % (
+                self.timestamp,
+                self.open,
+                self.close,
+                self.high,
+                self.low,
+                self.volume,
+                self.symbol,
+                self.created_time,
+                self.updated_time,
+            )
+        )

src/model/nse_processed_model.py

@@ -1,3 +1,5 @@
+import pandas as pd
+
 from datetime import datetime
 
 from src.app import server
@@ -6,6 +8,7 @@
 from src.repository.repository_response import generate_response
 
 postgres = server.get_postgres()
+engine = postgres.get_engine()
 session = postgres.get_session()
 log = Logger()
 
@@ -75,3 +78,15 @@ def delete_one(data_id):
         return generate_response(200, "delete", True, repr(nse_data_to_be_deleted))
     else:
         return generate_response(200, "delete", False, None)
+
+
+def find_dataframe_by_symbol(symbol):
+    return pd.read_sql_query(
+        "select * from nse_data_daily where nse_data_daily.symbol = %(symbol)s",
+        params={"symbol": symbol},
+        con=engine,
+    )
+
+
+def write_dataframe_in_sql(df):
+    df.to_sql("book_details", con=engine, if_exists="append", chunksize=1000)

src/repository/nse_repository.py

@@ -6,12 +6,14 @@
     NsePriceVolumeDeliverableData,
     NseDataCsvParseRequest,
     NseDataDeleteRequest,
+    ProcessNseDataRequest,
 )
 from src.services.nse_service import (
     create_nse_data_from_nse_pvd_data,
     create_nse_data_from_csv,
     delete_nse_data,
     upsert_nse_data_from_nse_pvd_data,
+    process_nse_data,
 )
 from src.util import parse_request_using_schema
 
@@ -46,3 +48,12 @@ def set_nse_pvd_data_csv():
     nse_data_csv_req = parse_request_using_schema(request, NseDataCsvParseRequest())
     service_resp = create_nse_data_from_csv(nse_data_csv_req)
     return make_response(jsonify(service_resp), 200)
+
+
+@api.route("/process", methods=["POST"])
+def process_nse_pvd_data():
+    process_nse_pvd_data_request = parse_request_using_schema(
+        request, ProcessNseDataRequest()
+    )
+    service_resp = process_nse_data(process_nse_pvd_data_request)
+    return make_response(jsonify(service_resp), service_resp["status"])

src/routes/nse/__init__.py

@@ -57,3 +57,10 @@ class NseDataCsvParseRequest(Schema):
 
     class Meta:
         strict = True
+
+
+class ProcessNseDataRequest(Schema):
+    symbol = fields.Str(required=True)
+
+    class Meta:
+        strict = True

src/routes/nse/schema.py

@@ -2,6 +2,7 @@
 
 import src.repository.nse_repository as nse_repo
 from src.common import Logger
+from src.lib.technical_indicators import moving_average, exponential_moving_average
 from src.routes.nse import NsePriceVolumeDeliverableData
 from src.routes.nse.transformer import get_nse_daily_data_model_from_nse_pvd_data
 from src.services.job_service import enqueue_job, create_batch_job
@@ -77,4 +78,23 @@ def parse_one_row_of_nse_data_csv(index, row):
         nse_data_dict["is_valid_data"] = False
 
     nse_data = nse_pvd_schema.dump(nse_data_dict)
-    upsert_nse_data_from_nse_pvd_data(nse_data)
+    return upsert_nse_data_from_nse_pvd_data(nse_data)
+
+
+def process_nse_data(process_nse_pvd_data_request):
+    symbol = process_nse_pvd_data_request["symbol"]
+    df = nse_repo.find_dataframe_by_symbol(symbol)
+    enqueue_job(process_and_insert_dataframe_in_database, "high", tuple([df]))
+    return {"success": True, "status": 200}
+
+
+def process_and_insert_dataframe_in_database(df):
+    log.info(f"process_and_insert_dataframe_in_database")
+    # TODO: schema
+    # TODO: add data in schema
+    df = moving_average(df, "close", 20)
+    df = exponential_moving_average(df, "close", 20)
+    df = df.fillna(0)
+    print(df)
+    # TODO: save to database
+    return 1