Fraud detection database API server

Question

Rather embarrassingly, I have been coding for some years now, however I feel that I still write code in a very basic way. It is hard to be certain, when many companies often give coding interviews for free labour, and the internet rarely sees software development from a business side but treats it too academically fussing over details that an ordinary company would not waste resources on for productivity.

I was given a task to write in Python by a fraud prevention company, which I completed and confirmed worked correctly, within their four to five hour deadline. They gave me a list of dummy transactions within a CSV file, and the following instructions:

1. Using Flask + any other python libraries design a simple API service that would be able to serve information in this CSV file.

2. Assume this service will be part of an additional BI + ML pipeline, so propose a number of aggregates that each API could provide when queried.

3. This is quite an open-ended task, so feel free to take the approach you feel is right and make some assumptions if they simplify the job.

Question: How could I improve my code and solution?

Question: Without worrying about my feelings, was this good work for 4 to 5 hours? I have a university degree in Computer Science, a paid industry year where I did software development (but I learnt very little as the company used a difficult to understand framework), and have tried hard to code since I was a young child.

I assume: The answers will vary from using a framework to reduce boilerplate code, dependency injection, test cases, using several py files, more comments, use switch instead of the numerous elif statements, error handling, proper logging rather than using print for status information, and better named variables. I guess the algorithm part of the code is poor, for example it would not search the BINs very efficiently, and a RAM database might have been a faster implementation given the high traffic this could potentially receive.

Please be harsh, because after so many years I am always doing the same basic things, I want to finally code the way industries, or startup companies, expect.

Here is the source code:

#################################################################################################
# Created by XXXXX YYYYY for ZZZZZ, use at your own risk
# Test the API with Post Man by sending JSON data:
# {
#   "auth": "amazingHorse",
#   "action": "suseptibleBINs"
#   "term": "additional information like search terms, else write 'empty'"
# }

# Setup instructions for first time users:
# Visual Studio Code
# pip3 install virtualenv
# .\env\Scripts\activate.bat
# pip3 flask flask-sqlalchemy
# pip3 install pandas
# python app.py
# !!! Warning: delete the xtransactions.sqlite file before each new deployment!!!!
#################################################################################################

from flask import Flask, request, jsonify
import sqlalchemy as db
import pandas as pd
from itertools import islice
import collections
import csv
import time
app = Flask(__name__)

adminPassword = "cyberExpert333" # Admin API password
userPassword = "amazingHorse" # User API password
suseptibleBINs = ["111","222","2a6f738d9ce1a8a9381aa775620fe5f0"] 
req_data = "" # Stores the JSON received from API

# For ordinary visitors
@app.route('/')
def hello_world():
    return 'Welcome, this is a private nonproduction system and provides API access to authorised users only'

# TODO admin access with special privs to drop tables etc...
@app.route('/api/admin', methods=['POST'])
def adminAccess():
    req_data = request.get_json()
    password = req_data['password']
    if(password != adminPassword):
        return "Authentication failure"
    else:
        return "Welcome administrator, please finish coding this..."

# Normal API access
@app.route('/api/user', methods=['POST'])
def userAccess():
    req_data = request.get_json()
    password = req_data['password'] # This function only checks for user password, admins have a seperate path at /api/admin
    if(password != userPassword):
        return "Authentication failure"
    else:
        action = req_data['action'] # What user would like to do
        term = req_data['term'] # Optional information, such as text to search for
        if(action == "BIN"):
            return binsFraudRisk()
        elif (action == "search"):
            return searchDatabase(term)
        elif (action == "mean"):
            return meanMaths(term)
        elif (action == "mode"):
            return modeMaths(term)
        elif (action == "median"):
            return medianMaths(term)
        elif (action == "max"):
            return maxMaths(term)
        elif (action == "min"):
            return minMaths(term)
        else:
            return "Error in your JSON data, please resend request"

# Boiler plate database code
def boilerPlateDatabase():
    engine = db.create_engine('sqlite:///xtransactions.sqlite')
    connection = engine.connect()
    metadata = db.MetaData()
    census = db.Table('trans', metadata, autoload=True, autoload_with=engine)
    query = db.select([census])
    ResultProxy = connection.execute(query)
    ResultSet = ResultProxy.fetchall()
    return ResultSet

# BINs suseptible to fraud
def binsFraudRisk():
    ResultSet = boilerPlateDatabase()
    foundRow = list()
    for row in ResultSet:
        for badBIN in suseptibleBINs:
            if badBIN == row[4]: 
                foundRow.append(row) # Add into foundRow array
    return str(foundRow) # All discovered rows which have suseptible BINs

# Search string - case sensitive!
def searchDatabase(searchTerm):
    ResultSet = boilerPlateDatabase()
    foundRow = list()
    for row in ResultSet:
        if (searchTerm == row[1] or searchTerm == row[2] or searchTerm == row[3] or searchTerm == row[4] or searchTerm == row[5] or searchTerm == row[6] or searchTerm == row[7] or searchTerm == row[8] or searchTerm == row[9]): 
            foundRow.append(row)
    return str(foundRow) # All rows which match the query

# Mean is the average
def meanMaths(searchTerm):
    ResultSet = boilerPlateDatabase()
    searchNumber = 0
    # 7
    if(searchTerm == "amount"):
        searchNumber = 7
    # 8
    elif(searchTerm == "currency"):
        searchNumber = 8
    else:
        return "Error in your JSON"

    # New array with just with numbers from amount or currency
    num = []
    for number in ResultSet:
        num.append(number[searchNumber])

    # Calculate the mean
    sum_num = 0
    for t in num:
        sum_num = sum_num + t           
    avg = sum_num / len(num)
    return str(avg)

# Mode is number that appears the most
def modeMaths(searchTerm):
    ResultSet = boilerPlateDatabase()
    searchNumber = 0
    # 7
    if(searchTerm == "amount"):
        searchNumber = 7
    # 8
    elif(searchTerm == "currency"):
        searchNumber = 8
    else:
        return "Error in your JSON"

    # New array with just with numbers from amount or currency
    num_list = []
    for number in ResultSet:
        num_list.append(number[searchNumber])

    # calculate the frequency of each item
    data = collections.Counter(num_list)
    data_list = dict(data)

    max_value = max(list(data.values()))
    mode_val = [num for num, freq in data_list.items() if freq == max_value]
    if len(mode_val) == len(num_list):
        return "Mode not found!"
    else:
        return("The Mode of the list is : " + ', '.join(map(str, mode_val)))

# Median is middle value
def medianMaths(searchTerm):
    ResultSet = boilerPlateDatabase()
    searchNumber = 0
    # 7
    if(searchTerm == "amount"):
        searchNumber = 7
    # 8
    elif(searchTerm == "currency"):
        searchNumber = 8
    else:
        return "Error in your JSON"

    # New array with just with numbers from amount or currency
    num_list = []
    for number in ResultSet:
        num_list.append(number[searchNumber])

    # Sort the list
    num_list.sort()
    # Finding the position of the median
    if len(num_list) % 2 == 0:
        first_median = num_list[len(num_list) // 2]
        second_median = num_list[len(num_list) // 2 - 1]
        median = (first_median + second_median) / 2
    else:
        median = num_list[len(num_list) // 2]
    return("The median is: " + str(median))

# Max is highest number
def maxMaths(searchTerm):
    ResultSet = boilerPlateDatabase()
    searchNumber = 0
    # 7
    if(searchTerm == "amount"):
        searchNumber = 7
    # 8
    elif(searchTerm == "currency"):
        searchNumber = 8
    else:
        return "Error in your JSON"

    # New array with just with numbers from amount or currency
    num_list = []
    for number in ResultSet:
        num_list.append(number[searchNumber])
    
    return 'Maximum: '+str(max(num_list))

# Min is smallest number
def minMaths(searchTerm):
    ResultSet = boilerPlateDatabase()
    searchNumber = 0
    # 7
    if(searchTerm == "amount"):
        searchNumber = 7
    # 8
    elif(searchTerm == "currency"):
        searchNumber = 8
    else:
        return "Error in your JSON"

    # New array with just with numbers from amount or currency
    num_list = []
    for number in ResultSet:
        num_list.append(number[searchNumber])

    return 'Minimum: '+str(min(num_list))

def databaseManagement():
    print("Creating new database")
    engine = db.create_engine('sqlite:///xtransactions.sqlite') # Create test.sqlite automatically
    connection = engine.connect()
    metadata = db.MetaData()
    trans = db.Table('trans', metadata,
        db.Column('Id', db.Integer(), unique=True), # Not in CSV file but makes life easier
        db.Column('t', db.String(), nullable=False),
        db.Column('tx_id', db.String(), nullable=False),
        db.Column('src_card', db.String(), nullable=False),
        db.Column('src_BIN', db.String(), nullable=False),
        db.Column('dst_card', db.String(), nullable=False),
        db.Column('dst_BIN', db.String(), nullable=False),
        db.Column('amount', db.Integer(), nullable=False),
        db.Column('currency', db.Integer(), nullable=False),
        db.Column('status', db.String(), nullable=False),
        )
    metadata.create_all(engine) # Creates the table

    # Inserting record one by one with a FOR loop
    print("Loading CSV into database")
    theIDCounter = 0 # Counter variable to assign a unique ID for DB
    with open('trx_sample.csv') as fd:
        for row in islice(csv.reader(fd), 2, None): # Skip line 0 (the disclaimer) and 1 (which is stuff like t, tx_id)
            query = db.insert(trans).values(Id=theIDCounter, t=row[0], tx_id=row[1], src_card=row[2], src_BIN=row[3], dst_card=row[4], dst_BIN=row[5], amount=row[6], currency=row[7], status=row[8])
            ResultProxy = connection.execute(query)
            theIDCounter = theIDCounter + 1
    results = connection.execute(db.select([trans])).fetchall()

if __name__ == '__main__':
    databaseManagement() # Loads CSV and puts everything into the DB
    app.run(debug=False) # If true then it causes server to restart which upsets DB

Answer 1

Your code doesn't look Pythonic as your code does not follow PEP 8.

• A lot of PEP 8 is to do with whitespace; two newlines above and below top level class and function definitions, two spaces in front of an inline comment, a space always after a comma, a single space either side of operators.
• You're using camelCase rather than snake_case for functions and variable names.
• You have a lot of unneeded parentheses making your code more dense.
• You're inconsistent in your string delimiters.

A couple of these issues and given that you have a function named hello_world makes me think you've clearly taken the Flask code from the web.

You have also used comments rather than docstrings, PEP 257, to document your code.

---

Your server API doesn't seem too great:

• All responses are 200 OK even in failure where some 400 codes would make sense.

• Your API doesn't have a common format to easily identify if a response was successful whether or not you're not using HTTP codes.

  Your error messages don't even have a common format or word.

---

In all your code at a short glance has a few red flags.

Answer 2

a long block of elif-s can be simplified by using a dictionary lookup. So instead of this:

if(action == "BIN"):
    return binsFraudRisk()
elif (action == "search"):
    return searchDatabase(term)
elif (action == "mean"):
    return meanMaths(term)
elif (action == "mode"):
    return modeMaths(term)
elif (action == "median"):
    return medianMaths(term)
elif (action == "max"):
    return maxMaths(term)
elif (action == "min"):
    return minMaths(term)
else:
    return "Error in your JSON data, please resend request"

you can define a dictionary with condition as a key + result as a value, assuming all functions expect the same parameters (ie. add term parameter to binsFraudRisk() even if it will be unused there)

action_resolver = {
    "BIN": binsFraudRisk,
    "search": searchDatabase,
    # ...keep adding other functions
}

# all if-elifs will shrink into this:
if action in action_resolver:
    return action_resolver[action](term)
else:
    return "Error in your JSON data, please resend request"
    

Answer 3

The other answers already address topics like code formatting or other things which are maybe not idiomatic. In this review, I want to focus on what I would expect if I was the interviewer and would review your solution.

1. Design a simple API

The first task ask you to design a simple API which serves some data from a CSV file. As an interviewer, I would want to see the following things:

• the API uses a common data format to provide the data to the end user, most likely JSON.
• The API is well behaved, e.g. it sets the correct Content-Type headers, and meaningful status codes.
• the interface of the API makes sense in the context of the application and is consistent.
• the API is well documented.

So a good first approach would be to look at the data you have at hand. From your database schema in the databaseManagement function, I assume you have a list of transactions in the CSV file. So the minimum API I would like to see is:

GET /transactions - returns all transactions in the CSV file in a JSON format.
GET /transactions/:id - returns the transaction with the given tx_id.

Bonus points, if you'd use something like openapi to specify the data types and operations of your API with this standard. This standard also allows you to provide the API documentation in an easy way.

Other functionalities that could be implemented on top would be for example filtering down the list of transactions using query parameters:

GET /transactions?currency=EUR - returns all transactions with currency equal to EUR.

Or you could add other resources to index by, which would be helpful for the second task:

GET /accounts - assuming a BIN is some sort of account ID, return all existing src_BINs.
GET /accounts/:src_bin - return some info about the given account (e.g. a balance).

Why?

In a professional environment it is extremely important to carefully design your APIs, because once they are out in the wild, you can (almost) never change them. Other tools will depend on the endpoints to behave in the documented way. I would want to see that the candidate takes this into consideration.

2. Propose a number of aggregates that each API could provide when queried

I think for this task it is mostly about the context of the interview you are in. This is a fraud detection company, and I think this should be reflected in your answer to this task.

I am not from this business, so no idea what would make sense, but if I were to do something, I'd do something like:

GET /accounts/:src_bin/statistics - returns min, max, mean of outgoing transactions of the given :src_bin.

or even something of higher level like:

GET /accounts/:src_bin/outliers- returns out of the ordinary transactions for the given :src_bin

Where you implement any outlier detection algorithm you find on the internet, to allow API users to find "suspicious" transactions.

Why?

This shows that you are not just another coding monkey, but instead that you are able to think in the business domain of the company you're going to work with, which is a very important skill for any software engineer.

Topics to learn about

You explicitly ask about advice on how to

[...] finally code the way industries, or startup companies, expect.

And I think there are a few skills you should acquire which will help you tackle tasks like this more easily:

• know the well-established standards. Other answers mentioned the pep8 code formatting convention, but also you should know about openapi and REST when you're working in any web related engineering position.
• if you introduce additional dependencies, there should be a good reason for it. I'm thinking about using SQLite as a database. I think it technically is a good choice, but you don't really use any of its features in your code. All the aggregates that you provide don't use any of SQL's aggregation functionality.
• know how to do API authentication. The task didn't ask for authentication, so I think the choice was to either don't do it at all, or do it right. Hardcoding a password and sending it in the request body is not a right way of doing it.

Answer 4

The code can be much simpler. It sounds like you already have a lot of ideas on potential improvements, so I would recommend trying those out more often and comparing the result of each one to how the code was before in order to learn from it.

• Your only database query is a query to fetch all rows. If you’re going to load a CSV into an SQLite database, it should be to run queries against.

• Python has built-in median calculation.

• if(searchTerm == "amount"): searchNumber = 7 elif(searchTerm == "currency"): searchNumber = 8 and the column extraction that follows is some very factor-out-able common logic.

• It’s probably better API design to give each operation its own route. Doing this will also force you to factor out authentication, which also improves the code.

• If the statistics aren’t changing at runtime, you can precompute them.

• Comparing passwords with == isn’t safe because of timing attacks, but it’s a toy example, and your interviewers probably won’t care/know about that.

• Pandas is unused.

• There are a lot of Python standard library, language features, and formatting expectations you should learn if you’re going to be using Python. (Of course, some of that can be on the job. But other things, like checking if an element is in a list, you should know can be done better no matter which language you’re familiar with.)

I hope to update this answer with more detail tomorrow. Anyway, here’s an idea of what the code can look like in the meantime:

import statistics
from dataclasses import dataclass


@dataclass
class Transaction:
    t: str
    tx_id: str
    src_card: str
    src_bin: str
    dst_card: str
    dst_bin: str
    amount: int
    currency: int   # you sure?
    status: str     # enum?


@app.route("/amounts/median")
def median():
    return {"median": statistics.median(t.amount for t in transactions)}


# ... (every route is that short) ...


# libraries can do this
def transaction_from_row(row):
    return Transaction(
        t=row[0],
        tx_id=row[1],
        src_card=row[2],
        src_bin=row[3],
        dst_card=row[4],
        dst_bin=row[5],
        amount=int(row[6]),
        currency=int(row[7]),
        status=row[8],
    )


if __name__ == "__main__":
    with open("trx_sample.csv") as f:
        # Skip line 0 (the disclaimer) and 1 (which is stuff like t, tx_id)
        transactions = [transaction_from_row(row) for row in islice(csv.reader(fd), 2)]

Answer 5

These observations are just about the code itself and not about your implementation

The vast majority of Python code bases adhere to the PEP8 style guide, this helps Python developers parse and understand each others code. Your code does not follow PEP8 guidelines for function and variable names. Function and variable names should be lowercase_with_underscores:

def bins_fraud_risk():
    result_set = boiler_plate_database()

You are wrapping expressions in your "if" statements with unnecessary parenthesis, remove them:

if search_term == "amount":

Concatenating strings with the "+" operator is frowned upon as it creates a new string every time which can lead to poor performance. String formatting is generally preferred

# Using an f-string
return f'The median is: {median}'

# Using .format()
return 'The median is: {}'.format(median)

Answer 6

Personally there is one thing that puts me off: the use of index values instead of field names. Consider this example:

# BINs suseptible to fraud
def binsFraudRisk():
    ResultSet = boilerPlateDatabase()
    foundRow = list()
    for row in ResultSet:
        for badBIN in suseptibleBINs:
            if badBIN == row[4]: 
                foundRow.append(row) # Add into foundRow array
    return str(foundRow) # All discovered rows which have suseptible BINs

If you have a SQL database then I don't understand why you do this stuff (row by row iteration to find bad BINs) when a simple SQL query would do the job. This is not efficient.

But the other problem is the reference to field #4. What if your table structure changes ? It is perfectly conceivable that a few more fields may be added/inserted in the future, and that the position of other fields will shift as a result. So you'll have to make plenty of adjustments in your code and there is always a risk that you will forget one line (or more). Then your code will probably continue to work but some conditions will no longer evaluate the way you intended. This is a good way to introduce bugs.

And using columns number is not intuitive at all.

Your code is full of hardcoded magic numbers too. If you can't avoid them, use constants.

You make repeated calls to boilerPlateDatabase() which in turns does this: ResultSet = ResultProxy.fetchall(), which is terribly inefficient and unnecessary. Just do regular SQL queries. I think that you didn't take the time yet to read up about SQLAlchemy (since it's the platform you've chosen). Python also supports SQLite out of the box so you could make queries easily, even without SQLAlchemy.

Or if you are going to load all records to memory do it once only, on startup. Because you are working on a static sample of data that is not going to change anyway. No need to reload it all every time.

My advice would be:

• if you are dealing with a SQL database, learn to use proper SQL
• make it a habit to measure the execution time of your code, here is a useful pointer: https://realpython.com/python-timer/ You can achieve massive gains by improving your code. The impact may not be very noticeable right now but when you run expensive code in a loop your users will be getting impatient behind their screens.

Pandas could have been an option but in the present case regular SQL should be more than enough.

But think of what would happen with a very large dataset. Your application will not scale and become slower (or crash) and loading the whole thing to memory could lead to performance issues. Just for kicks I would have a look at memory usage and CPU levels while the application is running. I found flaws in my own applications just by doing that. Like, a command-line application taking up 25% CPU or 50 Mb RAM => not looking hood.

In one word you need to learn how to write efficient code and not just code that does the work.

Last but not least: seems to me that the API should be implemented as a standalone class, thus in a separate file. It is distinct than the interface.

The comments in your code are not really helpful: I learn nothing about the functions by looking at them. What would help is two or three line explaining the logic and showing a data sample perhaps.