pylord/API-BFSI
0
1"""2RiskShield API Testing Client3Comprehensive testing utility for all endpoints4"""5 6import requests7import json8from datetime import datetime, timedelta9import random10from typing import List,Dict, Any11 12BASE_URL = "http://localhost:8000"13 14class RiskShieldClient:15 def __init__(self, base_url: str = BASE_URL):16 self.base_url = base_url17 self.session = requests.Session()18 19 def health_check(self) -> Dict[str, Any]:20 """Check API health"""21 response = self.session.get(f"{self.base_url}/api/health")22 return response.json()23 24 def register_user(self, full_name: str, email: str, password: str) -> Dict[str, Any]:25 """Register a new user"""26 payload = {27 "full_name": full_name,28 "email": email,29 "password": password30 }31 response = self.session.post(f"{self.base_url}/api/register", json=payload)32 return response.json()33 34 def login_user(self, email: str, password: str) -> Dict[str, Any]:35 """Login user"""36 payload = {37 "email": email,38 "password": password39 }40 response = self.session.post(f"{self.base_url}/api/login", json=payload)41 return response.json()42 43 def predict_transaction(self, 44 email: str,45 customer_id: str,46 transaction_id: str,47 transaction_datetime: str,48 transaction_amount: float,49 kyc_verified: int,50 account_age_days: int,51 channel_encoded: int) -> Dict[str, Any]:52 """Predict fraud for a transaction"""53 payload = {54 "email": email,55 "customer_id": customer_id,56 "transaction_id": transaction_id,57 "transaction_datetime": transaction_datetime,58 "transaction_amount": transaction_amount,59 "kyc_verified": kyc_verified,60 "account_age_days": account_age_days,61 "channel_encoded": channel_encoded62 }63 response = self.session.post(f"{self.base_url}/api/predict", json=payload)64 return response.json()65 66 def get_transactions(self, email: str) -> Dict[str, Any]:67 """Get transaction history for a user"""68 response = self.session.get(f"{self.base_url}/api/transactions/{email}")69 return response.json()70 71 def get_analytics(self) -> Dict[str, Any]:72 """Get analytics dashboard data"""73 response = self.session.get(f"{self.base_url}/api/analytics")74 return response.json()75 76 def get_metrics(self) -> Dict[str, Any]:77 """Get model metrics"""78 response = self.session.get(f"{self.base_url}/api/metrics")79 return response.json()80 81 def bulk_predict(self, email: str, transactions: List[Dict[str, Any]]) -> Dict[str, Any]:82 """Bulk predict multiple transactions"""83 payload = {84 "email": email,85 "transactions": transactions86 }87 response = self.session.post(f"{self.base_url}/api/bulk-predict", json=payload)88 return response.json()89 90def generate_test_transaction(email: str, customer_id: str, 91 scenario: str = "normal") -> Dict[str, Any]:92 """Generate test transaction data based on scenario"""93 94 transaction_id = f"TXN{random.randint(10000, 99999)}"95 now = datetime.now()96 97 scenarios = {98 "normal": {99 "transaction_amount": random.uniform(500, 5000),100 "kyc_verified": 1,101 "account_age_days": random.randint(100, 500),102 "channel_encoded": random.randint(0, 3),103 "hour": random.randint(9, 18)104 },105 "high_risk": {106 "transaction_amount": random.uniform(100000, 200000),107 "kyc_verified": 0,108 "account_age_days": random.randint(1, 10),109 "channel_encoded": 0,110 "hour": random.randint(22, 23)111 },112 "night_transaction": {113 "transaction_amount": random.uniform(60000, 90000),114 "kyc_verified": 1,115 "account_age_days": random.randint(50, 200),116 "channel_encoded": 1,117 "hour": random.randint(0, 5)118 },119 "weekend_high": {120 "transaction_amount": random.uniform(85000, 120000),121 "kyc_verified": 1,122 "account_age_days": random.randint(30, 100),123 "channel_encoded": 2,124 "hour": random.randint(10, 16)125 }126 }127 128 config = scenarios.get(scenario, scenarios["normal"])129 130 # Adjust datetime for scenario131 txn_time = now.replace(hour=config["hour"], minute=random.randint(0, 59))132 133 # For weekend scenario, adjust to Saturday or Sunday134 if scenario == "weekend_high":135 days_to_add = (5 - txn_time.weekday()) % 7136 txn_time = txn_time + timedelta(days=days_to_add)137 138 return {139 "email": email,140 "customer_id": customer_id,141 "transaction_id": transaction_id,142 "transaction_datetime": txn_time.strftime("%Y-%m-%d %H:%M:%S"),143 "transaction_amount": config["transaction_amount"],144 "kyc_verified": config["kyc_verified"],145 "account_age_days": config["account_age_days"],146 "channel_encoded": config["channel_encoded"]147 }148 149 150def run_comprehensive_test():151 """Run comprehensive API tests"""152 153 client = RiskShieldClient()154 155 print("=" * 60)156 print("š”ļø RiskShield API Comprehensive Test")157 print("=" * 60)158 159 # 1. Health Check160 print("\n1ļøā£ Health Check")161 print("-" * 60)162 health = client.health_check()163 print(json.dumps(health, indent=2))164 165 # 2. Register User166 print("\n2ļøā£ User Registration")167 print("-" * 60)168 test_email = f"test_{random.randint(1000, 9999)}@example.com"169 test_password = "TestPass123"170 171 register_response = client.register_user(172 full_name="Test User",173 email=test_email,174 password=test_password175 )176 print(json.dumps(register_response, indent=2))177 178 # 3. Login179 print("\n3ļøā£ User Login")180 print("-" * 60)181 login_response = client.login_user(test_email, test_password)182 print(json.dumps(login_response, indent=2))183 184 # 4. Test Different Transaction Scenarios185 print("\n4ļøā£ Transaction Predictions")186 print("-" * 60)187 188 scenarios = ["normal", "high_risk", "night_transaction", "weekend_high"]189 customer_id = f"CUST{random.randint(1000, 9999)}"190 191 for scenario in scenarios:192 print(f"\n š Testing: {scenario.upper().replace('_', ' ')}")193 print(" " + "-" * 56)194 195 txn_data = generate_test_transaction(test_email, customer_id, scenario)196 prediction = client.predict_transaction(**txn_data)197 198 if prediction.get("status") == "success":199 data = prediction["data"]200 print(f" ā Transaction ID: {txn_data['transaction_id']}")201 print(f" ā Amount: ā¹{txn_data['transaction_amount']:,.2f}")202 print(f" ā Combined Score: {data['combined_score']}")203 print(f" ā Is Fraud: {'Yes' if data['is_fraud'] else 'No'}")204 print(f" ā Rules Triggered: {len(data['rules_triggered'])}")205 if data['rules_triggered']:206 for rule in data['rules_triggered']:207 print(f" ā¢ {rule}")208 else:209 print(f" ā Error: {prediction.get('detail', 'Unknown error')}")210 211 # 5. Get Transaction History212 print("\n5ļøā£ Transaction History")213 print("-" * 60)214 history = client.get_transactions(test_email)215 if history.get("status") == "success":216 data = history["data"]217 print(f"Total Transactions: {data['total_transactions']}")218 print(f"User: {data['user_name']}")219 print(json.dumps(history, indent=2)[:500] + "...")220 221 # 6. Get Analytics222 print("\n6ļøā£ Analytics Dashboard")223 print("-" * 60)224 analytics = client.get_analytics()225 if analytics.get("status") == "success":226 kpis = analytics["data"]["kpis"]227 print(f"Total Transactions: {kpis['total_transactions']}")228 print(f"Fraud Detected: {kpis['fraud_detected']}")229 print(f"Accuracy Rate: {kpis['accuracy_rate']}%")230 print(f"Amount Protected: ā¹{kpis['amount_protected']:,.2f}")231 232 # 7. Get Model Metrics233 print("\n7ļøā£ Model Metrics")234 print("-" * 60)235 metrics = client.get_metrics()236 if metrics.get("status") == "success":237 model_metrics = metrics["data"]["metrics"]238 print(f"Accuracy: {model_metrics['accuracy']:.3f}")239 print(f"Precision: {model_metrics['precision']:.3f}")240 print(f"Recall: {model_metrics['recall']:.3f}")241 print(f"F1 Score: {model_metrics['f1_score']:.3f}")242 print(f"AUC-ROC: {model_metrics['auc_roc']:.3f}")243 244 print("\n" + "=" * 60)245 print("ā
Comprehensive Test Completed!")246 print("=" * 60)247 248 249def quick_fraud_test():250 """Quick test for high-risk fraud scenarios"""251 252 client = RiskShieldClient()253 254 print("\nšØ Quick Fraud Detection Test")255 print("=" * 60)256 257 # Use existing user or create new one258 test_email = "quicktest@example.com"259 test_password = "QuickTest123"260 261 try:262 client.register_user("Quick Test", test_email, test_password)263 except:264 pass # User might already exist265 266 # Test high-risk transaction267 txn_data = generate_test_transaction(test_email, "CUST9999", "high_risk")268 269 print(f"\nš Transaction Details:")270 print(f" Amount: ā¹{txn_data['transaction_amount']:,.2f}")271 print(f" Time: {txn_data['transaction_datetime']}")272 print(f" KYC: {'Verified' if txn_data['kyc_verified'] else 'Not Verified'}")273 print(f" Account Age: {txn_data['account_age_days']} days")274 275 result = client.predict_transaction(**txn_data)276 277 if result.get("status") == "success":278 data = result["data"]279 print(f"\nšÆ Prediction Results:")280 print(f" Risk Score: {data['combined_score']:.2%}")281 print(f" Fraud Status: {'FRAUDULENT' if data['is_fraud'] else 'LEGITIMATE'}")282 print(f" Model Score: {data['model_risk_score']:.2%}")283 print(f" Rule Score: {data['rule_score']:.2%}")284 285 if data['rules_triggered']:286 print(f"\nā ļø Rules Triggered:")287 for rule in data['rules_triggered']:288 print(f" ā¢ {rule}")289 290 print(f"\nš” Explanation:")291 print(f" {data['explanation'][:200]}...")292 293 print("\n" + "=" * 60)294 295def test_bulk_predict():296 """Test bulk prediction functionality"""297 298 client = RiskShieldClient()299 300 print("\n" + "=" * 60)301 print("š¦ Bulk Prediction Test")302 print("=" * 60)303 304 # Register test user305 test_email = f"bulk_test_{random.randint(1000, 9999)}@example.com"306 test_password = "BulkTest123"307 308 try:309 client.register_user("Bulk Test User", test_email, test_password)310 except:311 pass312 313 # Generate 50 test transactions314 transactions = []315 customer_id = f"CUST{random.randint(1000, 9999)}"316 317 for i in range(50):318 scenario = random.choice(["normal", "high_risk", "night_transaction", "weekend_high"])319 txn = generate_test_transaction(test_email, customer_id, scenario)320 # Remove email field as it's sent separately321 txn.pop("email", None)322 transactions.append(txn)323 324 print(f"\nš Testing bulk prediction with {len(transactions)} transactions...")325 326 # Make bulk prediction327 result = client.bulk_predict(test_email, transactions)328 329 if result.get("status") == "success":330 data = result["data"]331 print(f"\nā
Bulk Prediction Results:")332 print(f" Total Processed: {data['total_processed']}")333 print(f" Successful: {data['successful']}")334 print(f" Failed: {data['failed']}")335 print(f" Fraud Detected: {data['fraud_detected']}")336 print(f" Fraud Rate: {data['fraud_rate']}%")337 print(f" Processing Time: {data['processing_time_seconds']}s")338 print(f" Avg Time/Transaction: {data['avg_time_per_transaction_ms']}ms")339 340 # Show sample results341 print(f"\nš Sample Results (first 5):")342 for result in data['results'][:5]:343 status_icon = "ā" if result['status'] == "success" else "ā"344 fraud_icon = "šØ" if result['is_fraud'] else "ā
"345 print(f" {status_icon} {result['transaction_id']}: {fraud_icon} Risk={result['risk_score']:.2%}")346 else:347 print(f"\nā Error: {result.get('detail', 'Unknown error')}")348 349 print("\n" + "=" * 60)350 351if __name__ == "__main__":352 import sys353 354 if len(sys.argv) > 1:355 if sys.argv[1] == "quick":356 quick_fraud_test()357 elif sys.argv[1] == "bulk":358 test_bulk_predict()359 else:360 run_comprehensive_test()