不正検出データベースAPIサーバー

Question

恥ずかしいことに、私はここ数年コーディングを続けていますが、それでも非常に基本的な方法でコードを書いていると感じています。多くの企業が無料の労働のためにコーディングインタビューを行うことが多く、インターネットがビジネス側からソフトウェア開発を見ることはめったにありませんが、通常の企業が生産性のためにリソースを浪費しない詳細について学術的に混乱しすぎている場合、確信が持てません。

詐欺防止会社からPythonで書くタスクが与えられ、4〜5時間の期限内に完了し、正しく機能することを確認しました。彼らは私にCSVファイル内のダミートランザクションのリストと次の指示を与えました：

1. Flask +その他のPythonライブラリを使用して、このCSVファイルで情報を提供できるシンプルなAPIサービスを設計します。

2. このサービスは追加のBI + MLパイプラインの一部であると想定しているため、クエリ時に各APIが提供できるいくつかの集計を提案します。

3. これは非常に制限のないタスクなので、自分が正しいと思うアプローチを自由に採用し、作業が単純化される場合はいくつかの仮定を立ててください。

質問：コードとソリューションを改善するにはどうすればよいですか？

質問：私の気持ちを気にせずに、これは4〜5時間の良い仕事でしたか？私はコンピュータサイエンスの大学の学位を持っています。これは、ソフトウェア開発を行った有償の業界年です（ただし、会社が理解しにくいフレームワークを使用していたため、ほとんど学びませんでした）。幼い頃からコーディングに励んでいました。

：私は仮定の代わりに、多数のelif文、エラー処理、適切なログ機能ではなく、ステータスのプリントを使用しての答えはいくつかのPYファイルを使用して、定型的なコード、依存性注入、テストケースを減らすためのフレームワークを使用してから変化します、より多くのコメントが、使用スイッチ情報、およびより適切な名前の変数。コードのアルゴリズム部分は貧弱だと思います。たとえば、BINをあまり効率的に検索できず、RAMデータベースは、受信する可能性のあるトラフィックが多いため、より高速な実装であった可能性があります。

何年もの間、私はいつも同じ基本的なことをしているので、厳しくしてください。私は最終的に、業界やスタートアップ企業が期待する方法をコーディングしたいと思っています。

ソースコードは次のとおりです。

#################################################################################################
# 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

あなたのコードは従わないとしてあなたのコードはPython的に見えないPEP 8。

• PEP 8の多くは、空白に関係しています。トップレベルのクラスと関数の定義の上下に2つの改行、インラインコメントの前に2つのスペース、常にコンマの後にスペース、演算子の両側に1つのスペース。
• 関数や変数名camelCaseではなく、snake_caseを使用しています。
• 不要な括弧がたくさんあるため、コードの密度が高くなっています。
• 文字列の区切り文字に一貫性がありません。

これらの問題のいくつかと、という名前の関数hello_worldがあることを考えると、FlaskコードをWebから明確に取得したと思います。

また、コメントではなく、ドキュメンテーション文字列、使用しているPEP 257をあなたのコードを文書化するために、。

---

サーバーAPIはそれほど優れていないようです。

• 約400のコードが意味をなす障害が発生した場合でも、すべての応答は200OKです。

• APIには、HTTPコードを使用しているかどうかに関係なく、応答が成功したかどうかを簡単に識別するための共通の形式がありません。

  エラーメッセージには、一般的な形式や単語すらありません。

---

一目でわかるすべてのコードには、いくつかの危険信号があります。

Answer 2

elif-sの長いブロックは、辞書ルックアップを使用して簡略化できます。したがって、これの代わりに：

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"

すべての関数が同じパラメーターを期待していると仮定して、条件をキー+結果を値として持つ辞書を定義できます（つまり、そこで使用されない場合でもtermパラメーターを追加しますbinsFraudRisk()）

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

他の回答は、コードのフォーマットや慣用的ではないかもしれない他の事柄のようなトピックをすでに扱っています。このレビューでは、私がインタビュアーであり、あなたの解決策をレビューする場合に私が期待することに焦点を当てたいと思います。

1.シンプルなAPIを設計する

最初のタスクでは、CSVファイルからいくつかのデータを提供する単純なAPIを設計するように求められます。インタビュアーとして、私は次のことを見たいと思います。

• APIは、一般的なデータ形式を使用して、データをエンドユーザー（おそらくJSON）に提供します。
• APIは適切に動作します。たとえば、正しいContent-Typeヘッダーと意味のあるステータスコードを設定します。
• APIのインターフェースは、アプリケーションのコンテキストで意味があり、一貫性があります。
• APIは十分に文書化されています。

したがって、最初の良いアプローチは、手元にあるデータを確認することです。databaseManagement関数内のデータベーススキーマから、CSVファイルにトランザクションのリストがあると想定します。したがって、私が見たい最小のAPIは次のとおりです。

GET /transactions-CSVファイル内のすべてのトランザクションをJSON形式で返します。
GET /transactions/:id-指定されたtx_id。のトランザクションを返します。

ボーナスポイント。openapiのようなものを使用して、この標準でAPIのデータ型と操作を指定する場合。この標準では、APIドキュメントを簡単な方法で提供することもできます。

上に実装できる他の機能は、たとえば、クエリパラメータを使用してトランザクションのリストをフィルタリングすることです。

GET /transactions?currency=EUR -EURに等しい通貨ですべてのトランザクションを返します。

または、他のリソースをインデックスに追加することもできます。これは、2番目のタスクに役立ちます。

GET /accounts-BINが何らかのアカウントIDであると仮定して、既存のすべてのsrc_BINを返します。
GET /accounts/:src_bin-指定されたアカウントに関する情報（残高など）を返します。

どうして？

プロフェッショナルな環境では、APIを慎重に設計することが非常に重要です。これは、APIが公開されると、（ほとんど）変更できないためです。他のツールは、文書化された方法で動作するエンドポイントに依存します。候補者がこれを考慮に入れていることを確認したいと思います。

2.クエリ時に各APIが提供できる集計の数を提案します

このタスクでは、主にあなたが行っているインタビューのコンテキストに関するものだと思います。これは不正検出会社であり、これをこのタスクへの回答に反映させる必要があると思います。

私はこのビジネスの出身ではないので、何が意味があるのか​​わかりませんが、何かをする場合は、次のようなことをします。

GET /accounts/:src_bin/statistics-戻りmin、max、mean与えられたの出取引の:src_bin。

または次のようなより高いレベルのものですら：

GET /accounts/:src_bin/outliers-指定された通常のトランザクションからのリターン :src_bin

インターネット上で見つけた外れ値検出アルゴリズムを実装して、APIユーザーが「疑わしい」トランザクションを見つけられるようにします。

どうして？

これは、あなたが単なるコーディングモンキーではなく、一緒に仕事をする会社のビジネスドメインで考えることができることを示しています。これは、ソフトウェアエンジニアにとって非常に重要なスキルです。

学ぶべきトピック

あなたは明示的に方法についてのアドバイスについて尋ねます

[...]最後に、業界やスタートアップ企業が期待する方法をコーディングします。

そして、このようなタスクにもっと簡単に取り組むのに役立つ、習得すべきスキルがいくつかあると思います。

• 確立された基準を知っています。他の回答では、pep8コードのフォーマット規則について言及されていましたが、Web関連のエンジニアリング職で作業している場合は、openapiとRESTについても知っておく必要があります。
• 追加の依存関係を導入する場合は、それには正当な理由があるはずです。SQLiteをデータベースとして使うことを考えています。技術的には良い選択だと思いますが、コードでその機能を実際に使用することはありません。提供するすべての集計は、SQLの集計機能を使用しません。
• API認証を行う方法を知っています。タスクは認証を要求しなかったので、認証をまったく行わないか、正しく行うかのどちらかを選択したと思います。パスワードをハードコーディングしてリクエスト本文に送信することは、正しい方法ではありません。

Answer 4

コードははるかに単純にすることができます。改善の可能性についてはすでに多くのアイデアがあるようですので、それらをもっと頻繁に試して、それぞれの結果を以前のコードと比較して、そこから学ぶことをお勧めします。

• 唯一のデータベースクエリは、すべての行をフェッチするクエリです。CSVをSQLiteデータベースにロードする場合は、クエリを実行する必要があります。

• Pythonには中央値計算が組み込まれています。

• if(searchTerm == "amount"): searchNumber = 7 elif(searchTerm == "currency"): searchNumber = 8 次の列抽出は、非常に因数分解可能な一般的なロジックです。

• 各操作に独自のルートを与える方が、おそらくより良いAPI設計です。これを行うと、認証を除外する必要があり、コードも改善されます。

• 実行時に統計が変更されていない場合は、事前に計算できます。

• パスワードをと比較する==ことはタイミング攻撃のために安全ではありませんが、それはおもちゃの例であり、インタビュアーはおそらくそれを気にしない/知りません。

• パンダは未使用です。

• Pythonを使用する場合は、Pythonの標準ライブラリ、言語機能、およびフォーマットに関する期待がたくさんあります。（もちろん、その一部は仕事に使用できます。ただし、要素がリストに含まれているかどうかを確認するなど、他のことは、どの言語に精通していても、より適切に実行できることを知っておく必要があります。）

明日、この回答をより詳細に更新したいと思います。とにかく、その間にコードがどのように見えるかについての考えは次のとおりです。

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

これらの観察は、コード自体に関するものであり、実装に関するものではありません。

Pythonコードベースの大部分はPEP8スタイルガイドに準拠しています。これは、Python開発者がお互いのコードを解析して理解するのに役立ちます。コードは、関数名と変数名に関するPEP8ガイドラインに準拠していません。関数名と変数名は次のようにする必要がありますlowercase_with_underscores。

def bins_fraud_risk():
    result_set = boiler_plate_database()

「if」ステートメントの式を不要な括弧で囲んでいる場合は、それらを削除します。

if search_term == "amount":

文字列を「+」演算子で連結すると、パフォーマンスが低下する可能性があるたびに新しい文字列が作成されるため、嫌われます。文字列のフォーマットが一般的に好まれます

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

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

Answer 6

個人的に私を思いとどまらせることが1つあります。それは、フィールド名の代わりにインデックス値を使用することです。この例を考えてみましょう。

# 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

SQLデータベースがある場合、単純なSQLクエリがその仕事をするのに、なぜこのようなことをするのか（行ごとに繰り返して、悪いBINを見つける）はわかりません。これは効率的ではありません。

しかし、他の問題はフィールド＃4への参照です。テーブル構造が変更された場合はどうなりますか？将来、さらにいくつかのフィールドが追加/挿入される可能性があり、その結果、他のフィールドの位置がシフトすることは完全に考えられます。そのため、コードを十分に調整する必要があり、1行（またはそれ以上）を忘れるリスクが常にあります。その後、コードはおそらく機能し続けますが、一部の条件では意図した方法で評価されなくなります。これはバグを導入するための良い方法です。

また、列番号の使用は直感的ではありません。

あなたのコードもハードコードされたマジックナンバーでいっぱいです。それらを避けられない場合は、定数を使用してください。

あなたは繰り返し呼び出しを行い、boilerPlateDatabase()それが順番にこれを行います：ResultSet = ResultProxy.fetchall()これはひどく非効率的で不必要です。通常のSQLクエリを実行するだけです。SQLAlchemyについて読むのにまだ時間がかかっていなかったと思います（選択したプラットフォームであるため）。PythonはSQLiteもサポートしているため、SQLAlchemyがなくても簡単にクエリを実行できます。

または、すべてのレコードをメモリにロードする場合は、起動時に1回だけロードします。とにかく変更されないデータの静的サンプルで作業しているためです。毎回リロードする必要はありません。

私のアドバイスは次のとおりです。

• SQLデータベースを扱っている場合は、適切なSQLの使用方法を学びます
• コードの実行時間を測定することを習慣にしてください。ここに便利なポインタがあります。https://realpython.com/python-timer/コードを改善することで、大幅な利益を得ることができます。現時点では影響はそれほど目立たないかもしれませんが、高価なコードをループで実行すると、ユーザーは画面の後ろで焦ります。

パンダはオプションであった可能性がありますが、この場合、通常のSQLで十分です。

しかし、非常に大きなデータセットで何が起こるか考えてみてください。アプリケーションはスケーリングせず、遅くなる（またはクラッシュする）ため、すべてをメモリにロードするとパフォーマンスの問題が発生する可能性があります。キックのためだけに、アプリケーションの実行中のメモリ使用量とCPUレベルを確認します。それをするだけで、自分のアプリケーションに欠陥が見つかりました。同様に、25％のCPUまたは50MbのRAMを使用するコマンドラインアプリケーション=>見た目が悪い。

一言で言えば、あなたは仕事をするコードだけでなく、効率的なコードを書く方法を学ぶ必要があります。

最後になりましたが、APIはスタンドアロンクラスとして、つまり別のファイルに実装する必要があるように思われます。それはインターフェースとは異なります。

あなたのコードのコメントはあまり役に立ちません：私はそれらを見ても関数について何も学びません。役立つのは、ロジックを説明し、おそらくデータサンプルを示す2行または3行です。