CSV parsing with multiple symbols

Author: pritam001

console.sql

@@ -6,19 +6,20 @@ CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE;
 -- We start by creating a regular SQL table
 
 CREATE TABLE nse_data_daily (
-  timestamp TIMESTAMP PRIMARY KEY NOT NULL,
+  timestamp TIMESTAMP NOT NULL,
   open DOUBLE PRECISION  NOT NULL,
   close    DOUBLE PRECISION  NOT NULL,
   high    DOUBLE PRECISION  NOT NULL,
   low    DOUBLE PRECISION  NOT NULL,
   volume    DOUBLE PRECISION  NOT NULL,
   symbol VARCHAR(20) NOT NULL,
   created_time TIMESTAMP NOT NULL,
-  updated_time TIMESTAMP NOT NULL
+  updated_time TIMESTAMP NOT NULL,
+  CONSTRAINT timestamp_symbol PRIMARY KEY(timestamp, symbol)
 );
 
 -- This creates a hypertable that is partitioned by time
---   using the values in the `time` column.
+-- using the values in the `time` column.
 
 SELECT create_hypertable('nse_data_daily', 'timestamp');
 

src/repository/nse_repository.py

@@ -26,7 +26,14 @@ def upsert_one(data):
     if not isinstance(data, NseDailyDataModel):
         raise RuntimeError("Model error")
 
-    nse_data = session.query(NseDailyDataModel).get(data.timestamp)
+    nse_data = (
+        session.query(NseDailyDataModel)
+        .filter(
+            NseDailyDataModel.timestamp == data.timestamp,
+            NseDailyDataModel.symbol == data.symbol,
+        )
+        .first()
+    )
     if nse_data is not None:
         update_params = {
             "timestamp": data.timestamp,
@@ -39,10 +46,18 @@ def upsert_one(data):
             "updated_time": datetime.utcnow(),
         }
         session.query(NseDailyDataModel).filter(
-            NseDailyDataModel.timestamp == data.timestamp
+            NseDailyDataModel.timestamp == data.timestamp,
+            NseDailyDataModel.symbol == data.symbol,
         ).update(update_params)
         session.commit()
-        updated_nse_data = session.query(NseDailyDataModel).get(data.timestamp)
+        updated_nse_data = (
+            session.query(NseDailyDataModel)
+            .filter(
+                NseDailyDataModel.timestamp == data.timestamp,
+                NseDailyDataModel.symbol == data.symbol,
+            )
+            .first()
+        )
         log.info(f"nse_repository : update_one : updated_nse_data = {updated_nse_data}")
         return generate_response(200, "update", True, repr(updated_nse_data))
     else:

src/routes/nse/schema.py

@@ -27,6 +27,7 @@ class NsePriceVolumeDeliverableData(Schema):
     no_of_trades = fields.Int(required=True)
     deliverable_qty = fields.Int(required=True)
     percent_daily_qty_to_traded_qty = fields.Float(required=True)
+    is_valid_data = fields.Bool(missing=True)
 
     @post_dump
     def date_to_string(self, output, **kwargs):

src/services/nse_service.py

@@ -65,6 +65,16 @@ def parse_one_row_of_nse_data_csv(index, row):
         "no_of_trades": row["No. of Trades"],
         "deliverable_qty": row["Deliverable Qty"],
         "percent_daily_qty_to_traded_qty": row["% Dly Qt to Traded Qty"],
+        "is_valid_data": True,
     }
+
+    # manual fixes on csv data
+    if nse_data_dict["deliverable_qty"] == "-":
+        nse_data_dict["deliverable_qty"] = 0
+        nse_data_dict["is_valid_data"] = False
+    if nse_data_dict["percent_daily_qty_to_traded_qty"] == "-":
+        nse_data_dict["percent_daily_qty_to_traded_qty"] = 0
+        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)