Add full-stack BTC portfolio web app

Multi-user FastAPI + React app with JWT auth, SQLite storage, and
CoinGecko price integration. Dockerized with docker-compose.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Jonathan
2026-03-23 22:15:40 +01:00
parent 84679639ef
commit 3907414742
27 changed files with 859 additions and 0 deletions
+28
View File
@@ -0,0 +1,28 @@
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
SECRET_KEY = "change-me-in-production-use-a-long-random-string"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # 1 day
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def create_access_token(data: dict) -> str:
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode["exp"] = expire
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def decode_token(token: str) -> dict:
return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+22
View File
@@ -0,0 +1,22 @@
import os
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./btc_portfolio.db")
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
+29
View File
@@ -0,0 +1,29 @@
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError
from sqlalchemy.orm import Session
from .auth import decode_token
from .database import get_db
from . import models
bearer_scheme = HTTPBearer()
def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
db: Session = Depends(get_db),
) -> models.User:
token = credentials.credentials
try:
payload = decode_token(token)
username: str = payload.get("sub")
if username is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
except JWTError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
user = db.query(models.User).filter(models.User.username == username).first()
if user is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
return user
+26
View File
@@ -0,0 +1,26 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from .database import engine, Base
from .routes import users, purchases, stats
Base.metadata.create_all(bind=engine)
app = FastAPI(title="BTC Portfolio API")
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(users.router)
app.include_router(purchases.router)
app.include_router(stats.router)
@app.get("/")
def root():
return {"message": "BTC Portfolio API"}
+26
View File
@@ -0,0 +1,26 @@
from sqlalchemy import Column, Integer, String, Float, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
from .database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True, nullable=False)
password = Column(String, nullable=False)
purchases = relationship("Purchase", back_populates="owner", cascade="all, delete")
class Purchase(Base):
__tablename__ = "purchases"
id = Column(Integer, primary_key=True, index=True)
amount_eur = Column(Float, nullable=False)
price_eur = Column(Float, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
owner = relationship("User", back_populates="purchases")
@@ -0,0 +1,67 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from pydantic import BaseModel
from typing import List
from datetime import datetime
from ..database import get_db
from .. import models
from ..dependencies import get_current_user
router = APIRouter()
class PurchaseCreate(BaseModel):
amount_eur: float
price_eur: float
class PurchaseOut(BaseModel):
id: int
amount_eur: float
price_eur: float
created_at: datetime
class Config:
from_attributes = True
@router.get("/purchases", response_model=List[PurchaseOut])
def list_purchases(
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user),
):
return db.query(models.Purchase).filter(models.Purchase.user_id == current_user.id).all()
@router.post("/purchases", response_model=PurchaseOut, status_code=status.HTTP_201_CREATED)
def add_purchase(
purchase_in: PurchaseCreate,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user),
):
purchase = models.Purchase(
amount_eur=purchase_in.amount_eur,
price_eur=purchase_in.price_eur,
user_id=current_user.id,
)
db.add(purchase)
db.commit()
db.refresh(purchase)
return purchase
@router.delete("/purchases/{purchase_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_purchase(
purchase_id: int,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user),
):
purchase = db.query(models.Purchase).filter(
models.Purchase.id == purchase_id,
models.Purchase.user_id == current_user.id,
).first()
if not purchase:
raise HTTPException(status_code=404, detail="Purchase not found")
db.delete(purchase)
db.commit()
+33
View File
@@ -0,0 +1,33 @@
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from ..database import get_db
from .. import models
from ..dependencies import get_current_user
from ..services.btc import get_btc_price_eur
router = APIRouter()
@router.get("/stats")
def get_stats(
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user),
):
purchases = db.query(models.Purchase).filter(models.Purchase.user_id == current_user.id).all()
total_invested = sum(p.amount_eur for p in purchases)
total_btc = sum(p.amount_eur / p.price_eur for p in purchases) if purchases else 0.0
average_price = total_invested / total_btc if total_btc > 0 else 0.0
current_price = get_btc_price_eur()
portfolio_value = total_btc * current_price
profit_loss = portfolio_value - total_invested
return {
"total_invested": round(total_invested, 2),
"total_btc": round(total_btc, 8),
"average_price": round(average_price, 2),
"current_price": round(current_price, 2),
"portfolio_value": round(portfolio_value, 2),
"profit_loss": round(profit_loss, 2),
}
+42
View File
@@ -0,0 +1,42 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from pydantic import BaseModel
from ..database import get_db
from .. import models
from ..auth import hash_password, verify_password, create_access_token
router = APIRouter()
class UserCreate(BaseModel):
username: str
password: str
class Token(BaseModel):
access_token: str
token_type: str
@router.post("/register", status_code=status.HTTP_201_CREATED)
def register(user_in: UserCreate, db: Session = Depends(get_db)):
existing = db.query(models.User).filter(models.User.username == user_in.username).first()
if existing:
raise HTTPException(status_code=400, detail="Username already taken")
user = models.User(
username=user_in.username,
password=hash_password(user_in.password),
)
db.add(user)
db.commit()
return {"message": "User created"}
@router.post("/login", response_model=Token)
def login(user_in: UserCreate, db: Session = Depends(get_db)):
user = db.query(models.User).filter(models.User.username == user_in.username).first()
if not user or not verify_password(user_in.password, user.password):
raise HTTPException(status_code=401, detail="Invalid credentials")
token = create_access_token({"sub": user.username})
return {"access_token": token, "token_type": "bearer"}
+14
View File
@@ -0,0 +1,14 @@
import requests
def get_btc_price_eur() -> float:
try:
resp = requests.get(
"https://api.coingecko.com/api/v3/simple/price",
params={"ids": "bitcoin", "vs_currencies": "eur"},
timeout=10,
)
resp.raise_for_status()
return float(resp.json()["bitcoin"]["eur"])
except Exception:
return 0.0