I have a requirement to write a Python 3.6 script which should be able to detect following case within a time period and be able to create a text file report that lists the condition violated and transaction that caused the violation:
- Redeeming over 3 time on same stamp card within 10 mins
The required table in database for storing user and stamp card details has following structure-
Here the 'to_sate' column with values 2,3 and 4 means a card is redeemed.
Now, I have written following script to fulfill above requirement and it is working as expected. In my local environment it is taking 2.7 minutes to complete the process for 36000 records-
def insert_data(csv_import_path, csv_export_path):
import time
from multiprocessing import Pool
import multiprocessing
import pandas as pd
import pymysql
# Connect to the database
engine = pymysql.connect(host='localhost',
user='admin',
password='MY_PASSWORD',
db='MY_DB',
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor)
df = pd.read_sql(
"SELECT stamps_record_id, user_id, stamp_card_id, stamp_time, merchant_id, merchant_store_id FROM rmsdb.user_stamping_records where to_state in (2,3,4) order by stamp_card_id",
engine)
df.to_csv(csv_import_path)
df = pd.read_csv(csv_import_path, index_col=["stamps_record_id"])
unique_users = df.user_id.unique()
df["stamp_time"] = pd.to_datetime(df["stamp_time"])
num_processes = multiprocessing.cpu_count()
s_time = time.time()
with Pool(num_processes) as p:
final_df = pd.DataFrame()
for i in range(0, len(unique_users)):
user = unique_users[i]
new_df = df[df.user_id == user]
sid = new_df.stamp_card_id.unique()
for i in sid:
fdf = new_df[new_df.stamp_card_id == i]
# len(fdf) can be user given value
if len(fdf) > 3:
for i in range(0, len(fdf)):
g = (fdf.iloc[i:i + 3])
if len(g) >= 3:
x = (g.tail(1).stamp_time.values - g.head(1).stamp_time.values).astype("timedelta64[s]")
if x[0].astype(int) < 600:
final_df = final_df.append(g)
e_time = time.time() - s_time
# final_df.drop_duplicates(keep="first").to_csv("C:\\Users\\rahul.khanna\\Desktop\\user_stamping_records_frauds.csv", index=False)
final_df.drop_duplicates(keep="first").to_csv(csv_export_path, index=False)
print("Total Time taken is: " + str(e_time / 60) + " minutes.")
if __name__ == '__main__':
insert_data("C:\\Users\\hitman\\Desktop\\user_stamping_records_import.csv", "C:\\Users\\hitman\\Desktop\\user_stamping_records_frauds.csv")
I have converted the df to dictionary for a sample-
34198: '2018-10-13 16:48:03', 34199: '2018-10-13 16:48:03', 34200: '2018-10-13 16:48:03', 34201: '2018-10-13 16:48:03', 34202: '2018-10-13 16:48:03', 34203: '2018-10-13 16:48:03', 34204: '2018-10-13 16:48:03', 34205: '2018-10-13 16:48:03', 34206: '2018-10-13 16:48:03', 34207: '2018-10-13 16:48:03', 34208: '2018-10-13 16:48:03'
Before moving my script to production, I need your suggestion for any improvement in my code.
Can anyone please have a look on my code and let me know any improvement in it?
Let me know also if I forgot to include any required information here.