diff --git a/.env b/.env new file mode 100644 index 0000000..d464550 --- /dev/null +++ b/.env @@ -0,0 +1,9 @@ +MONGODB_URL = mongodb://ssd2658:ssd2658@host.docker.internal:27017/portfolio?authSource=admin + +# Kafka +KAFKA_BOOTSTRAP_SERVERS = host.docker.internal:9093 +KAFKA_SECURITY_PROTOCOL = PLAINTEXT +KAFKA_SASL_MECHANISM = PLAIN +KAFKA_SASL_JAAS_CONFIG = #org.apache.kafka.common.security.plain.PlainLoginModule required username="kafkaUser" password="kafkaPassword123!"; + +AM_DOCUMENT_PROCESSOR_MAX_RETRIES = 15 \ No newline at end of file diff --git a/.github/workflows/docker-build.yml b/.github/workflows/docker-build.yml new file mode 100644 index 0000000..5c29299 --- /dev/null +++ b/.github/workflows/docker-build.yml @@ -0,0 +1,105 @@ +name: Docker Build and Push + +on: + push: + branches: + - 'main' + - 'develop' + - 'feature/**' + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }}/am-document-processor + +jobs: + build-and-push: + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + security-events: write + + steps: + - uses: actions/checkout@v3 + + - name: Set up JDK 17 + uses: actions/setup-java@v3 + with: + java-version: '17' + distribution: 'temurin' + cache: maven + + # Copy settings.xml from repo + - name: Setup Maven settings.xml + run: | + mkdir -p ~/.m2 + cp settings.xml ~/.m2/settings.xml + + # Build with Maven and cache dependencies + - name: Build with Maven + run: mvn clean package -DskipTests + env: + GITHUB_PACKAGES_USERNAME: ${{ github.actor }} + GITHUB_PACKAGES_TOKEN: ${{ github.token }} + + # Verify the build output exists + - name: Verify build output + run: | + echo "Maven build output:" + ls -la target/ + + # Set up Docker Buildx for multi-architecture builds + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Log in to GitHub Container Registry + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ github.token }} + + - name: Extract Docker metadata + id: meta + uses: docker/metadata-action@v5 + with: + images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} + tags: | + type=raw,value=v${{ github.run_id }} + type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }} + type=raw,value=develop,enable=${{ github.ref == 'refs/heads/develop' }} + type=ref,event=branch,prefix=feature-,enable=${{ startsWith(github.ref, 'refs/heads/feature/') }} + + # Build and push Docker image using the pre-built JAR + - name: Build and push Docker image + uses: docker/build-push-action@v5 + with: + context: . + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + platforms: linux/amd64,linux/arm64 + cache-from: type=gha + cache-to: type=gha,mode=max + + - name: Save image tag to artifact + run: | + echo "v${{ github.run_id }}" > image-tag.txt + + - name: Upload image tag as artifact + uses: actions/upload-artifact@v4 + with: + name: image-tag + path: image-tag.txt + retention-days: 1 + + # Trigger deployment workflow + deploy: + needs: build-and-push + permissions: + contents: read + id-token: write + uses: ./.github/workflows/unified-deployment.yml + with: + image_tag: v${{ github.run_id }} + secrets: inherit \ No newline at end of file diff --git a/.github/workflows/manual-deployment.yml b/.github/workflows/manual-deployment.yml new file mode 100644 index 0000000..37d8b85 --- /dev/null +++ b/.github/workflows/manual-deployment.yml @@ -0,0 +1,119 @@ +name: Manual AKS Deployment + +on: + workflow_dispatch: + inputs: + environment: + description: 'Environment to deploy to' + required: true + default: 'preprod' + type: choice + options: + - preprod + - prod + - feature/* + image_tag: + description: 'Docker image tag to deploy' + required: true + type: string + + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }}/am-document-processor + HELM_CHART_PATH: ./helm/am-document-processor + +jobs: + deploy: + name: Deploy to AKS + runs-on: ubuntu-latest + permissions: + contents: read + id-token: write + + steps: + # Checkout repository + # Checkout repository at the specified ref + - name: Checkout repository + uses: actions/checkout@v4 + + # Set environment-specific variables + - name: Set environment variables + id: set-vars + run: | + # Set environment-specific variables + if [[ "${{ github.event.inputs.environment }}" == "prod" ]]; then + echo "RESOURCE_GROUP=am-investing" >> $GITHUB_ENV + echo "CLUSTER_NAME=am-np-west" >> $GITHUB_ENV + echo "NAMESPACE=dev" >> $GITHUB_ENV + echo "Deploying to PRODUCTION environment" + elif [[ "${{ github.event.inputs.environment }}" == "feature" ]]; then + echo "RESOURCE_GROUP=am-investing" >> $GITHUB_ENV + echo "CLUSTER_NAME=am-np-west" >> $GITHUB_ENV + echo "NAMESPACE=dev" >> $GITHUB_ENV + echo "Deploying to FEATURE environment" + else + echo "RESOURCE_GROUP=am-investing" >> $GITHUB_ENV + echo "CLUSTER_NAME=am-np-west" >> $GITHUB_ENV + echo "NAMESPACE=dev" >> $GITHUB_ENV + echo "Deploying to PRE-PRODUCTION environment" + fi + + # Set image tag from input + echo "IMAGE_TAG=${{ github.event.inputs.image_tag }}" >> $GITHUB_ENV + + # Login to Azure + - name: Login to Azure + uses: azure/login@v1 + with: + creds: ${{ secrets.AZURE_CREDENTIALS }} + + # Set AKS context + - name: Set AKS context + uses: azure/aks-set-context@v3 + with: + resource-group: ${{ env.RESOURCE_GROUP }} + cluster-name: ${{ env.CLUSTER_NAME }} + + # Setup Helm + - name: Setup Helm + uses: azure/setup-helm@v3 + with: + version: 'v3.12.3' + + # Deploy to AKS using Helm + - name: Deploy to AKS + run: | + # Create namespace if it doesn't exist + kubectl create namespace ${{ env.NAMESPACE }} --dry-run=client -o yaml | kubectl apply -f - + + echo "Deploying image: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ env.IMAGE_TAG }}" + echo "Environment: ${{ github.event.inputs.environment }}" + echo "Resource Group: ${{ env.RESOURCE_GROUP }}" + echo "Cluster: ${{ env.CLUSTER_NAME }}" + echo "Namespace: ${{ env.NAMESPACE }}" + + # Set values file based on environment + if [[ "${{ github.event.inputs.environment }}" == "feature" ]]; then + # For feature branches, use preprod values as base + VALUES_FILE="${{ env.HELM_CHART_PATH }}/values/preprod.yaml" + else + # For standard environments, use the corresponding values file + VALUES_FILE="${{ env.HELM_CHART_PATH }}/values/${{ github.event.inputs.environment }}.yaml" + fi + + # Deploy using Helm with the appropriate values file + helm upgrade --install am-document-processor ${{ env.HELM_CHART_PATH }} \ + --namespace ${{ env.NAMESPACE }} \ + -f $VALUES_FILE \ + --set image.repository=${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} \ + --set image.tag=${{ env.IMAGE_TAG }} \ + --set deployment.version="${{ env.IMAGE_TAG }}" \ + --wait --timeout 5m + + # Verify deployment + - name: Verify deployment + run: | + kubectl rollout status deployment/am-document-processor -n ${{ env.NAMESPACE }} + echo "Deployment to ${{ github.event.inputs.environment }} completed successfully!" + echo "Deployment to ${{ github.event.inputs.environment }} completed successfully!" diff --git a/.github/workflows/unified-deployment.yml b/.github/workflows/unified-deployment.yml new file mode 100644 index 0000000..981fe86 --- /dev/null +++ b/.github/workflows/unified-deployment.yml @@ -0,0 +1,136 @@ +name: Deployment + +on: + # Manual deployment with environment selection + workflow_dispatch: + inputs: + environment: + description: 'Environment to deploy to' + required: true + default: 'preprod' + type: choice + options: + - preprod + - prod + image_tag: + description: 'Docker image tag to deploy (leave empty for latest build)' + required: false + type: string + + # Called by docker-build workflow + workflow_call: + inputs: + image_tag: + description: 'Docker image tag to deploy' + required: true + type: string + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }}/am-document-processor + HELM_CHART_PATH: ./helm/am-document-processor + IMAGE_TAG: ${{ github.event_name == 'workflow_dispatch' && github.event.inputs.image_tag || inputs.image_tag }} + +jobs: + # Deploy to the selected environment + deploy: + environment: ${{ github.event_name == 'workflow_dispatch' && github.event.inputs.environment || 'preprod' }} + name: Deploy to Environment + runs-on: ubuntu-latest + permissions: + contents: read + id-token: write + + env: + # Set environment based on workflow_dispatch input or default to preprod for workflow_call + ENVIRONMENT: ${{ github.event_name == 'workflow_dispatch' && github.event.inputs.environment || 'preprod' }} + + steps: + # Checkout repository + - name: Checkout repository + uses: actions/checkout@v4 + + # Set environment-specific variables + - name: Set environment variables + id: set-vars + run: | + # Set environment-specific variables + if [[ "${{ env.ENVIRONMENT }}" == "prod" ]]; then + echo "RESOURCE_GROUP=am-investing" >> $GITHUB_ENV + echo "CLUSTER_NAME=am-np-west" >> $GITHUB_ENV + echo "NAMESPACE=dev" >> $GITHUB_ENV + echo "Deploying to PRODUCTION environment" + else + echo "RESOURCE_GROUP=am-investing" >> $GITHUB_ENV + echo "CLUSTER_NAME=am-np-west" >> $GITHUB_ENV + echo "NAMESPACE=dev" >> $GITHUB_ENV + echo "Deploying to PRE-PRODUCTION environment" + fi + + # Set image tag from input + echo "IMAGE_TAG=${{ env.IMAGE_TAG }}" >> $GITHUB_ENV + + # Login to Azure + - name: Login to Azure + uses: azure/login@v1 + with: + creds: ${{ secrets.AZURE_CREDENTIALS }} + + # Set AKS context + - name: Set AKS context + uses: azure/aks-set-context@v3 + with: + resource-group: ${{ env.RESOURCE_GROUP }} + cluster-name: ${{ env.CLUSTER_NAME }} + + # Setup Helm + - name: Setup Helm + uses: azure/setup-helm@v3 + with: + version: '3.12.3' # Removed 'v' prefix for stability + + # Check and remove Helm locks if any exist + - name: Check and remove Helm locks + run: | + # Check if there's a release with pending operations + if kubectl get secret -n ${{ env.NAMESPACE }} | grep -q "sh.helm.release.v1.am-document-processor"; then + echo "Found existing Helm release, checking for locks..." + # Get all Helm secrets and check for pending operations + HELM_SECRETS=$(kubectl get secrets -n ${{ env.NAMESPACE }} -l owner=helm -o name) + for secret in $HELM_SECRETS; do + if [[ $secret == *"am-document-processor"* ]]; then + echo "Examining $secret for locks" + # Check if the secret contains a lock + LOCK_STATUS=$(kubectl get $secret -n ${{ env.NAMESPACE }} -o jsonpath='{.metadata.labels.status}') + if [[ $LOCK_STATUS == "pending-install" || $LOCK_STATUS == "pending-upgrade" || $LOCK_STATUS == "pending-rollback" ]]; then + echo "Found lock in $secret with status $LOCK_STATUS, removing..." + kubectl delete $secret -n ${{ env.NAMESPACE }} + echo "Lock removed" + fi + fi + done + else + echo "No existing Helm release found, proceeding with installation" + fi + + # Deploy to AKS using Helm + - name: Deploy to AKS + run: | + # Create namespace if it doesn't exist + kubectl create namespace ${{ env.NAMESPACE }} --dry-run=client -o yaml | kubectl apply -f - + + # Echo image and environment details for logging + echo "Deploying image: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ env.IMAGE_TAG }}" + echo "Environment: ${{ env.ENVIRONMENT }}" + echo "Resource Group: ${{ env.RESOURCE_GROUP }}" + echo "Cluster: ${{ env.CLUSTER_NAME }}" + echo "Namespace: ${{ env.NAMESPACE }}" + + # Deploy using Helm with the appropriate values file + helm upgrade --install am-document-processor ${{ env.HELM_CHART_PATH }} \ + --namespace ${{ env.NAMESPACE }} \ + -f ${{ env.HELM_CHART_PATH }}/values/${{ env.ENVIRONMENT }}.yaml \ + --set image.repository=${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} \ + --set image.tag=${{ env.IMAGE_TAG }} \ + --set deployment.version="${{ env.IMAGE_TAG }}" \ + --wait --timeout 5m \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..ce42369 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,28 @@ +# Single stage runtime image - uses pre-built JAR +FROM eclipse-temurin:17-jre-jammy + +WORKDIR /app + +# Copy the pre-built JAR file (will be copied in GitHub Actions) +COPY target/am-processor-*.jar app.jar + +# Install curl for healthcheck +RUN apt-get update && \ + apt-get install -y curl && \ + rm -rf /var/lib/apt/lists/* && \ + # Set timezone + ln -sf /usr/share/zoneinfo/Asia/Kolkata /etc/localtime + +# Set environment variables +ENV SPRING_PROFILES_ACTIVE=docker +ENV TZ=Asia/Kolkata + +# Expose the application port +EXPOSE 8080 + +# Health check +HEALTHCHECK --interval=30s --timeout=3s --start-period=60s --retries=3 \ + CMD curl -f http://localhost:8080/actuator/health || exit 1 + +# Run the application +ENTRYPOINT ["java", "-jar", "app.jar"] diff --git a/docker-compose.yml b/docker-compose.yml index 6759fb8..f7050e2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,33 +1,42 @@ version: '3.8' -services: - mongodb: - image: mongo:latest - container_name: mongodb - restart: always - ports: - - "27017:27017" +services: + am-document-processor-service: + container_name: am-document-processor-service + build: + context: . + dockerfile: Dockerfile + env_file: + - .env environment: - MONGO_INITDB_ROOT_USERNAME: admin - MONGO_INITDB_ROOT_PASSWORD: password - volumes: - - mongodb_data:/data/db - - mongo-express: - image: mongo-express:latest - container_name: mongo-express - restart: always + KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_BOOTSTRAP_SERVERS} + MONGODB_URL: ${MONGODB_URL} + + entrypoint: ["/bin/sh", "-c"] + command: + - | + echo "=== AM Document Processor Service Environment Variables ===" + echo "Database URL: ${MONGODB_URL}" + echo "Kafka Servers: ${KAFKA_BOOTSTRAP_SERVERS}" + echo "=== Starting AM Document Processor Service ===" + java -jar app.jar ports: - - "8081:8081" - environment: - ME_CONFIG_MONGODB_ADMINUSERNAME: admin - ME_CONFIG_MONGODB_ADMINPASSWORD: password - ME_CONFIG_MONGODB_SERVER: mongodb - ME_CONFIG_BASICAUTH_USERNAME: admin - ME_CONFIG_BASICAUTH_PASSWORD: password - depends_on: - - mongodb + - "8070:8080" + networks: + - market-data-network + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8080/actuator/health"] + interval: 30s + timeout: 3s + retries: ${AM_DOCUMENT_PROCESSOR_MAX_RETRIES} + restart: unless-stopped volumes: - mongodb_data: + postgres_data: + influxdb_data: + grafana_data: + +networks: + market-data-network: + driver: bridge diff --git a/helm/README.md b/helm/README.md new file mode 100644 index 0000000..c49e070 --- /dev/null +++ b/helm/README.md @@ -0,0 +1,147 @@ +# Market Data Service Helm Chart + +This Helm chart deploys the Market Data Service and its dependencies to AKS (Azure Kubernetes Service). + +## Prerequisites + +- Kubernetes 1.19+ +- Helm 3.0+ +- Azure Kubernetes Service (AKS) +- Nginx Ingress Controller +- Cert-Manager (for TLS) + +## Features + +- Automated deployment of Market Data Service with optimized configurations +- Built-in retry mechanism with configurable parameters +- Comprehensive metrics collection via Prometheus and InfluxDB +- Grafana dashboards for monitoring +- Horizontal Pod Autoscaling based on CPU and Memory +- Persistent storage for all stateful components +- TLS support via cert-manager + +## Installation + +1. Add required Helm repositories: +```bash +helm repo add bitnami https://charts.bitnami.com/bitnami +helm repo add influxdata https://helm.influxdata.com/ +helm repo add grafana https://grafana.github.io/helm-charts +helm repo add prometheus-community https://prometheus-community.github.io/helm-charts +helm repo update +``` + +2. Create a values override file (e.g., `custom-values.yaml`) and update the following: + - Image repository and credentials + - Domain name in ingress configuration + - Database credentials + - InfluxDB organization and token + - Kafka configuration if needed + +3. Install the chart: +```bash +helm install market-data ./market-data -f custom-values.yaml -n market-data --create-namespace +``` + +## Configuration + +### Critical Settings + +1. Market Data Processing: +```yaml +config: + marketData: + maxRetries: 3 # Maximum retry attempts for API calls + retryDelayMs: 1000 # Base delay between retries + threadPoolSize: 5 # Thread pool size for parallel processing + threadQueueCapacity: 10 # Queue capacity for pending tasks + maxAgeMinutes: 15 # Maximum age of market data +``` + +2. Resource Allocation: +```yaml +resources: + limits: + cpu: 1000m + memory: 1Gi + requests: + cpu: 500m + memory: 512Mi +``` + +3. Autoscaling: +```yaml +autoscaling: + enabled: true + minReplicas: 1 + maxReplicas: 3 + targetCPUUtilizationPercentage: 80 + targetMemoryUtilizationPercentage: 80 +``` + +### Dependencies + +1. PostgreSQL: +- Persistent storage: 10Gi +- Optimized for time-series data +- Configurable credentials + +2. Kafka: +- 3 replicas for high availability +- Zookeeper cluster included +- JMX metrics enabled +- Persistent storage: 10Gi + +3. InfluxDB: +- Prometheus endpoint enabled +- 30-day data retention +- Persistent storage: 10Gi + +4. Grafana: +- Auto-configured datasources +- Persistent storage: 5Gi +- Pre-configured dashboards + +5. Prometheus: +- 30-day metrics retention +- Persistent storage: 10Gi +- AlertManager included + +## Health Monitoring + +The service includes comprehensive health checks: +- Liveness probe: `/actuator/health/liveness` +- Readiness probe: `/actuator/health/readiness` +- Metrics endpoint: `/actuator/prometheus` + +## Upgrading + +To upgrade the deployment: +```bash +helm upgrade market-data ./market-data -f custom-values.yaml -n market-data +``` + +## Uninstallation + +To remove the deployment: +```bash +helm uninstall market-data -n market-data +``` + +## Troubleshooting + +1. Check pod status: +```bash +kubectl get pods -n market-data +``` + +2. View pod logs: +```bash +kubectl logs -f deployment/market-data -n market-data +``` + +3. Check service health: +```bash +kubectl port-forward svc/market-data 8084:8084 -n market-data +curl http://localhost:8084/actuator/health +``` diff --git a/helm/am-document-processor/Chart.yaml b/helm/am-document-processor/Chart.yaml new file mode 100644 index 0000000..ea6ba3c --- /dev/null +++ b/helm/am-document-processor/Chart.yaml @@ -0,0 +1,8 @@ +apiVersion: v2 +name: am-document-processor +description: AM Document Processor Service Helm Chart +type: application +version: 0.1.0 +appVersion: "1.0.0" +maintainers: + - name: AM Portfolio Team diff --git a/helm/am-document-processor/templates/_helpers.tpl b/helm/am-document-processor/templates/_helpers.tpl new file mode 100644 index 0000000..a3f6ba6 --- /dev/null +++ b/helm/am-document-processor/templates/_helpers.tpl @@ -0,0 +1,42 @@ +{{/* +Expand the name of the chart. +*/}} +{{- define "am-document-processor.name" -}} +{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Create a default fully qualified app name. +*/}} +{{- define "am-document-processor.fullname" -}} +{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Common labels +*/}} +{{- define "am-document-processor.labels" -}} +app.kubernetes.io/name: {{ include "am-document-processor.name" . }} +app.kubernetes.io/instance: {{ .Release.Name }} +app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} +app.kubernetes.io/managed-by: {{ .Release.Service }} +{{- end }} + +{{/* +Selector labels +*/}} +{{- define "am-document-processor.selectorLabels" -}} +app.kubernetes.io/name: {{ include "am-document-processor.name" . }} +app.kubernetes.io/instance: {{ .Release.Name }} +{{- end }} + +{{/* +Infrastructure service names +*/}} +{{- define "am-document-processor.postgresql.fullname" -}} +{{- .Values.postgresql.fullname }} +{{- end }} + +{{- define "am-document-processor.kafka.fullname" -}} +{{- .Values.kafka.bootstrapServers }} +{{- end }} diff --git a/helm/am-document-processor/templates/config.yaml b/helm/am-document-processor/templates/config.yaml new file mode 100644 index 0000000..fae3a37 --- /dev/null +++ b/helm/am-document-processor/templates/config.yaml @@ -0,0 +1,32 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: {{ include "am-document-processor.fullname" . }}-config + labels: + {{- include "am-document-processor.labels" . | nindent 4 }} +data: + # Spring Profile + SPRING_PROFILES_ACTIVE: "{{ .Values.environment }}" + TZ: "Asia/Kolkata" + + # Database Configuration + POSTGRES_URL: "jdbc:postgresql://{{ .Values.postgresql.fullname }}:{{ .Values.postgresql.primary.service.port }}" + POSTGRES_DATABASE: "{{ .Values.postgresql.auth.database }}" + # Vault Secret Paths for PostgreSQL + POSTGRES_SECRET_PATH: "{{ .Values.vault.secretPaths.database }}" + + # Note: The following environment variables are injected by Vault Agent and don't need to be defined here: + # - Kafka: KAFKA_* variables + + # Document Processing Configuration + DOC_PROC_MAX_RETRIES: "{{ .Values.config.documentProcessing.maxRetries }}" + DOC_PROC_RETRY_DELAY_MS: "{{ .Values.config.documentProcessing.retryDelayMs }}" + DOC_PROC_THREAD_POOL_SIZE: "{{ .Values.config.documentProcessing.threadPoolSize }}" + DOC_PROC_THREAD_QUEUE_CAPACITY: "{{ .Values.config.documentProcessing.threadQueueCapacity }}" + DOC_PROC_STORAGE_PROVIDER: "{{ .Values.config.documentProcessing.storageProvider }}" + + # Metrics Configuration + MANAGEMENT_ENDPOINTS_WEB_EXPOSURE_INCLUDE: "health,metrics,prometheus" + MANAGEMENT_ENDPOINT_HEALTH_SHOW_DETAILS: "always" + MANAGEMENT_METRICS_TAGS_APPLICATION: "document-processor-{{ .Values.environment }}" + MANAGEMENT_METRICS_EXPORT_PROMETHEUS_ENABLED: "{{ .Values.monitoring.prometheus.scrape }}" diff --git a/helm/am-document-processor/templates/deployment.yaml b/helm/am-document-processor/templates/deployment.yaml new file mode 100644 index 0000000..6b36c63 --- /dev/null +++ b/helm/am-document-processor/templates/deployment.yaml @@ -0,0 +1,88 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ include "am-document-processor.fullname" . }} + labels: + {{- include "am-document-processor.labels" . | nindent 4 }} +spec: + replicas: 1 + selector: + matchLabels: + {{- include "am-document-processor.selectorLabels" . | nindent 6 }} + template: + metadata: + annotations: + # Prometheus annotations for metrics collection + prometheus.io/scrape: "{{ .Values.monitoring.prometheus.scrape }}" + prometheus.io/path: "{{ .Values.monitoring.prometheus.path }}" + prometheus.io/port: "{{ .Values.monitoring.prometheus.port }}" + # Add checksum annotations for config changes + checksum/config: {{ include (print $.Template.BasePath "/config.yaml") . | sha256sum }} + # Vault Agent Injector annotations + vault.hashicorp.com/agent-inject: "true" + vault.hashicorp.com/auth-config-audience: "vault" # Adding audience as per warning in logs + vault.hashicorp.com/auth-config-issuer: "https://kubernetes.default.svc.cluster.local" + vault.hashicorp.com/auth-config-disable-iss-validation: "true" + vault.hashicorp.com/auth-config-disable-local-ca-jwt: "true" + vault.hashicorp.com/role: "dev-role" + vault.hashicorp.com/agent-pre-populate-only: "true" + vault.hashicorp.com/agent-inject-secret-kafka: "kv/preprod/kafka" + vault.hashicorp.com/agent-inject-template-kafka: | + {{`{{- with secret "kv/preprod/kafka" -}} + KAFKA_BOOTSTRAP_SERVERS="{{ .Data.data.kafkabootstrapserver }}" + KAFKA_SECURITY_PROTOCOL="{{ .Data.data.securityprotocol }}" + KAFKA_SASL_MECHANISM="{{ .Data.data.sslmechanism }}" + KAFKA_SASL_JAAS_CONFIG="{{ .Data.data.sasljaasconfig }}" + KAFKA_ZOOKEEPER_CONNECT="{{ .Data.data.zookeeperconnect }}" + KAFKA_CONSUMER_GROUP_ID="{{ .Data.data.consumerGroupId }}" + KAFKA_CONSUMER_AUTO_OFFSET_RESET="{{ .Data.data.autooffsetreset }}" + {{- end -}}`}} + + vault.hashicorp.com/agent-inject-secret-mongodb: "kv/preprod/database/mongo" + vault.hashicorp.com/agent-inject-template-mongodb: | + {{`{{- with secret "kv/preprod/database/mongo" -}} + MONGODB_DATABASE="{{ .Data.data.portfoliodatabase }}" + MONGODB_URI= {{ .Data.data.uri }}/{{ .Data.data.portfoliodatabase }}?authSource={{ .Data.data.authsource }} + {{- end -}}`}} + {{- with .Values.podAnnotations }} + {{- toYaml . | nindent 8 }} + {{- end }} + labels: + {{- include "am-document-processor.selectorLabels" . | nindent 8 }} + spec: + serviceAccountName: vault-auth + imagePullSecrets: + - name: ghcr-secret + containers: + - name: {{ .Chart.Name }} + image: "ghcr.io/am-portfolio/am-document-processor/am-document-processor:{{ .Values.image.tag | default .Chart.AppVersion }}" + imagePullPolicy: Always + # Load environment variables from ConfigMap + envFrom: + - configMapRef: + name: {{ include "am-document-processor.fullname" . }}-config + + ports: + - name: http + containerPort: 8080 + protocol: TCP + # Health checks with retry-based thresholds + livenessProbe: + httpGet: + path: /actuator/health/liveness + port: http + initialDelaySeconds: 60 + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: {{ .Values.config.documentProcessing.maxRetries }} + readinessProbe: + httpGet: + path: /actuator/health/readiness + port: http + initialDelaySeconds: 30 + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: {{ .Values.config.documentProcessing.maxRetries }} + # Resource limits from values file + resources: + {{- toYaml .Values.resources | nindent 12 }} diff --git a/helm/am-document-processor/templates/service.yaml b/helm/am-document-processor/templates/service.yaml new file mode 100644 index 0000000..7a6d224 --- /dev/null +++ b/helm/am-document-processor/templates/service.yaml @@ -0,0 +1,15 @@ +apiVersion: v1 +kind: Service +metadata: + name: {{ include "am-document-processor.fullname" . }} + labels: + {{- include "am-document-processor.labels" . | nindent 4 }} +spec: + type: {{ .Values.service.type }} + ports: + - port: {{ .Values.service.port }} + targetPort: 8080 + protocol: TCP + name: http + selector: + {{- include "am-document-processor.selectorLabels" . | nindent 4 }} diff --git a/helm/am-document-processor/values/preprod.yaml b/helm/am-document-processor/values/preprod.yaml new file mode 100644 index 0000000..360631b --- /dev/null +++ b/helm/am-document-processor/values/preprod.yaml @@ -0,0 +1,60 @@ +# Pre-production Environment Configuration +environment: preprod + +# Application Configuration +config: + documentProcessing: + maxRetries: 3 + retryDelayMs: 2000 + threadPoolSize: 4 + threadQueueCapacity: 20 + storageProvider: "s3" + +# Service Configuration +service: + type: ClusterIP + port: 8080 + +# Ingress Configuration +ingress: + enabled: true + ipAddress: 20.59.109.241 + host: "document-processor.am.local" + annotations: + kubernetes.io/ingress.class: nginx + +# Resource Configuration +resources: + limits: + cpu: 300m + memory: 512Mi + requests: + cpu: 200m + memory: 256Mi + +postgresql: + enabled: true + fullname: host.docker.internal + auth: + database: document_processor_db + username: postgres + password: password + primary: + service: + port: 5456 + +# Vault Configuration +vault: + enabled: true + secretPaths: + database: "vault:kv/data/preprod/database" + api: "vault:kv/data/preprod/api" + basepath: "vault:kv/data/preprod" + +# Monitoring Configuration +monitoring: + enabled: true + prometheus: + scrape: true + port: "8080" + path: "/actuator/prometheus" diff --git a/helm/am-document-processor/values/prod.yaml b/helm/am-document-processor/values/prod.yaml new file mode 100644 index 0000000..3b3fd6d --- /dev/null +++ b/helm/am-document-processor/values/prod.yaml @@ -0,0 +1,102 @@ +# Production Environment Configuration +environment: prod + +# Application Configuration +config: + documentProcessing: + maxRetries: ${DOC_MAX_RETRIES} + retryDelayMs: ${DOC_RETRY_DELAY_MS} + threadPoolSize: ${DOC_THREAD_POOL_SIZE} + threadQueueCapacity: ${DOC_THREAD_QUEUE_CAPACITY} + storageProvider: "s3" + +# Pod Configuration +replicaCount: 3 +podAnnotations: + prometheus.io/scrape: "true" + prometheus.io/port: "8080" + prometheus.io/path: "/actuator/prometheus" + fluentbit.io/parser: "java" + +# Pod Security Context +podSecurityContext: + runAsNonRoot: true + runAsUser: 1000 + fsGroup: 2000 + +# Container Security Context +securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + +# Service Configuration +service: + type: ClusterIP + port: 8080 + +# Resource Configuration +resources: + limits: + cpu: 2000m + memory: 2Gi + requests: + cpu: 1000m + memory: 1Gi + +# Infrastructure Configuration +postgresql: + enabled: true + fullname: am-doc-postgresql + auth: + database: document_processor_db + username: ${POSTGRES_USERNAME} + password: ${POSTGRES_PASSWORD} + primary: + service: + port: 5432 + resources: + limits: + cpu: 2000m + memory: 2Gi + requests: + cpu: 1000m + memory: 1Gi + +kafka: + enabled: true + bootstrapServers: am-prod-kafka:9092 + config: + consumerGroupId: am-document-processor-prod + autoOffsetReset: earliest + topics: + documentUpload: am-document-upload-prod + documentProcessed: am-document-processed-prod + replicationFactor: 3 + numPartitions: 6 + +# Autoscaling Configuration +autoscaling: + enabled: true + minReplicas: 3 + maxReplicas: 10 + targetCPUUtilizationPercentage: 70 + targetMemoryUtilizationPercentage: 70 + +# Ingress Configuration +ingress: + enabled: true + className: nginx + annotations: + kubernetes.io/ingress.class: nginx + cert-manager.io/cluster-issuer: letsencrypt-prod + hosts: + - host: docproc.aminvestment.com + paths: + - path: / + pathType: Prefix + tls: + - secretName: docproc-tls + hosts: + - docproc.aminvestment.com diff --git a/pom.xml b/pom.xml index 7bdcd59..0a41f00 100644 --- a/pom.xml +++ b/pom.xml @@ -32,11 +32,9 @@ 3.4.1 - 1.2.4-SNAPSHOT - 1.2.4-SNAPSHOT - - - 12.8.1.jre11 + 1.2.10-SNAPSHOT + 1.2.10-SNAPSHOT + 1.2.10-SNAPSHOT 2.15.2 @@ -57,6 +55,20 @@ 2.3.0 + + + github-investment + GitHub AM Common Investment Service Apache Maven Packages + https://maven.pkg.github.com/AM-Portfolio/am-common-investment-service + + + + github-am-common + GitHub AM Common Data Model Apache Maven Packages + https://maven.pkg.github.com/AM-Portfolio/am-common-data-parent + + + @@ -71,10 +83,6 @@ org.springframework.boot spring-boot-starter-actuator - - org.springframework.boot - spring-boot-starter-data-jpa - org.springframework.boot spring-boot-devtools @@ -94,21 +102,17 @@ ${am.common.data.model.version} + + com.am.common + am-common-data-service + ${am.common.data.service.version} + + org.springframework.boot spring-boot-starter-data-mongodb - - com.microsoft.sqlserver - mssql-jdbc - ${mssql.jdbc.version} - - - org.postgresql - postgresql - runtime - diff --git a/settings.xml b/settings.xml new file mode 100644 index 0000000..051a8f9 --- /dev/null +++ b/settings.xml @@ -0,0 +1,18 @@ + + + + + github-investment + ${env.GITHUB_PACKAGES_USERNAME} + ${env.GITHUB_PACKAGES_TOKEN} + + + github-am-common + ${env.GITHUB_PACKAGES_USERNAME} + ${env.GITHUB_PACKAGES_TOKEN} + + + diff --git a/src/main/java/org/am/mypotrfolio/DocumentProcessingApplication.java b/src/main/java/org/am/mypotrfolio/DocumentProcessingApplication.java index 24c6b2d..a3a5805 100644 --- a/src/main/java/org/am/mypotrfolio/DocumentProcessingApplication.java +++ b/src/main/java/org/am/mypotrfolio/DocumentProcessingApplication.java @@ -5,6 +5,7 @@ import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.ComponentScans; +import org.springframework.data.mongodb.repository.config.EnableMongoRepositories; import lombok.extern.slf4j.Slf4j; @@ -16,6 +17,7 @@ @ComponentScan("org.am.mypotrfolio.service.mapper"), @ComponentScan("com.am.common.amcommondata.mapper") }) +@EnableMongoRepositories(basePackages = "com.am.common.amcommondata.repository") public class DocumentProcessingApplication { public static void main(String[] args) { diff --git a/src/main/java/org/am/mypotrfolio/config/ApiAutoConfiguration.java b/src/main/java/org/am/mypotrfolio/config/ApiAutoConfiguration.java new file mode 100644 index 0000000..f25a20f --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/config/ApiAutoConfiguration.java @@ -0,0 +1,29 @@ +package org.am.mypotrfolio.config; + +import org.springframework.boot.autoconfigure.AutoConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Bean; +import org.springframework.web.servlet.config.annotation.CorsRegistry; +import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; + +@AutoConfiguration +@ConditionalOnProperty(prefix = "am.trade.api", name = "enabled", havingValue = "true", matchIfMissing = true) +@ComponentScan(basePackages = "org.am.mypotrfolio.api") +public class ApiAutoConfiguration { + + @Bean + public WebMvcConfigurer corsConfigurer() { + return new WebMvcConfigurer() { + @Override + public void addCorsMappings(CorsRegistry registry) { + registry.addMapping("/api/**") + .allowedOrigins("http://localhost:3000") // Frontend URL + .allowedMethods("GET", "POST", "PUT", "DELETE", "OPTIONS") + .allowedHeaders("*") + .allowCredentials(true) + .maxAge(3600); // 1 hour max age + } + }; + } +} diff --git a/src/main/java/org/am/mypotrfolio/config/LotSizeConfig.java b/src/main/java/org/am/mypotrfolio/config/LotSizeConfig.java new file mode 100644 index 0000000..32a99de --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/config/LotSizeConfig.java @@ -0,0 +1,103 @@ +package org.am.mypotrfolio.config; + +import java.math.BigDecimal; +import java.time.LocalDate; +import java.util.HashMap; +import java.util.Map; +import java.util.NavigableMap; +import java.util.TreeMap; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.context.annotation.Configuration; +import org.springframework.stereotype.Component; +import org.springframework.beans.factory.InitializingBean; + +import lombok.Data; + +/** + * Configuration for lot sizes that can change over time + * The configuration is loaded from application.yml + */ +@Data +@Component +@Configuration +@ConfigurationProperties(prefix = "trade.lot-sizes") +public class LotSizeConfig implements InitializingBean { + + private Map> indices = new HashMap<>(); + private Map defaultLotSizes = new HashMap<>(); + + // Cache for efficient lookup + private Map> lotSizeCache = new HashMap<>(); + + /** + * Initialize the cache after properties are set + */ + @Override + public void afterPropertiesSet() throws Exception { + lotSizeCache.clear(); + + // Process each index configuration + for (Map.Entry> indexEntry : indices.entrySet()) { + String indexName = indexEntry.getKey(); + Map dateToLotSizeMap = indexEntry.getValue(); + + NavigableMap timelineMap = new TreeMap<>(); + + // Convert string dates to LocalDate for the timeline + for (Map.Entry entry : dateToLotSizeMap.entrySet()) { + LocalDate effectiveDate = LocalDate.parse(entry.getKey()); + timelineMap.put(effectiveDate, entry.getValue()); + } + + lotSizeCache.put(indexName.toUpperCase(), timelineMap); + } + } + + /** + * Get the lot size for a given index on a specific date + * + * @param indexName the name of the index (e.g., NIFTY, BANKNIFTY) + * @param date the date for which to get the lot size + * @return the lot size applicable on the given date, or default if not found + */ + public BigDecimal getLotSize(String indexName, LocalDate date) { + if (indexName == null || date == null) { + return BigDecimal.ONE; + } + + // Ensure cache is initialized + if (lotSizeCache.isEmpty()) { + try { + afterPropertiesSet(); + } catch (Exception e) { + // Handle exception or rethrow as runtime exception + throw new RuntimeException("Failed to initialize lot size cache", e); + } + } + + NavigableMap timeline = lotSizeCache.get(indexName.toUpperCase()); + + if (timeline != null && !timeline.isEmpty()) { + // Get the entry with the greatest key less than or equal to the given date + Map.Entry entry = timeline.floorEntry(date); + if (entry != null) { + return entry.getValue(); + } + } + + // If no specific lot size found, return the default for this index + BigDecimal defaultLotSize = defaultLotSizes.get(indexName.toUpperCase()); + return defaultLotSize != null ? defaultLotSize : BigDecimal.ONE; + } + + /** + * Get the current lot size for a given index + * + * @param indexName the name of the index (e.g., NIFTY, BANKNIFTY) + * @return the current lot size, or default if not found + */ + public BigDecimal getCurrentLotSize(String indexName) { + return getLotSize(indexName, LocalDate.now()); + } +} diff --git a/src/main/java/org/am/mypotrfolio/config/MongoConfig.java b/src/main/java/org/am/mypotrfolio/config/MongoConfig.java deleted file mode 100644 index 0f06cf3..0000000 --- a/src/main/java/org/am/mypotrfolio/config/MongoConfig.java +++ /dev/null @@ -1,40 +0,0 @@ -package org.am.mypotrfolio.config; - -import org.am.mypotrfolio.config.properties.PersistenceProperties; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.data.mongodb.config.AbstractMongoClientConfiguration; -import org.springframework.data.mongodb.core.MongoTemplate; -import org.springframework.data.mongodb.repository.config.EnableMongoRepositories; -import org.springframework.lang.NonNull; - -import com.mongodb.client.MongoClient; -import com.mongodb.client.MongoClients; - -import lombok.RequiredArgsConstructor; - -@Configuration -@EnableMongoRepositories(basePackages = "com.am.common.amcommondata.repository.*") -@RequiredArgsConstructor -public class MongoConfig extends AbstractMongoClientConfiguration { - - private final PersistenceProperties persistenceProperties; - - @Override - @NonNull - protected String getDatabaseName() { - return persistenceProperties.getMongodb().getDatabase(); - } - - @Override - @Bean - @NonNull - public MongoClient mongoClient() { - return MongoClients.create(persistenceProperties.getMongodb().getUri()); - } - - @Bean - public MongoTemplate mongoTemplate() { - return new MongoTemplate(mongoClient(), getDatabaseName()); - } -} \ No newline at end of file diff --git a/src/main/java/org/am/mypotrfolio/config/properties/PersistenceProperties.java b/src/main/java/org/am/mypotrfolio/config/properties/PersistenceProperties.java deleted file mode 100644 index 04dd96d..0000000 --- a/src/main/java/org/am/mypotrfolio/config/properties/PersistenceProperties.java +++ /dev/null @@ -1,19 +0,0 @@ -package org.am.mypotrfolio.config.properties; - -import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.context.annotation.Configuration; - -import lombok.Data; - -@Data -@Configuration -@ConfigurationProperties(prefix = "app.persistence") -public class PersistenceProperties { - private MongoProperties mongodb; - - @Data - public static class MongoProperties { - private String uri; - private String database; - } -} diff --git a/src/main/java/org/am/mypotrfolio/controller/DocumentProcessorController.java b/src/main/java/org/am/mypotrfolio/controller/DocumentProcessorController.java index 531789e..594a40e 100644 --- a/src/main/java/org/am/mypotrfolio/controller/DocumentProcessorController.java +++ b/src/main/java/org/am/mypotrfolio/controller/DocumentProcessorController.java @@ -44,8 +44,12 @@ public ResponseEntity processDocument( @Parameter(description = "Portfolio document file to process", required = true) @RequestParam("file") MultipartFile file, @Parameter(description = "Type of document being processed", required = true) - @RequestParam("documentType") DocumentType documentType) { - return ResponseEntity.ok(documentProcessorService.processDocument(file, documentType)); + @RequestParam("documentType") DocumentType documentType, + @Parameter(description = "Portfolio ID", required = false) + @RequestParam(value = "portfolioId") String portfolioId, + @Parameter(description = "User ID", required = false) + @RequestParam(value = "userId") String userId) { + return ResponseEntity.ok(documentProcessorService.processDocument(file, documentType, portfolioId, userId)); } @Operation( @@ -63,8 +67,12 @@ public ResponseEntity> processBatchDocuments( @Parameter(description = "List of portfolio document files to process", required = true) @RequestParam("files") List files, @Parameter(description = "Type of documents being processed", required = true) - @RequestParam("documentType") DocumentType documentType) { - return ResponseEntity.ok(documentProcessorService.processBatchDocuments(files, documentType)); + @RequestParam("documentType") DocumentType documentType, + @Parameter(description = "Portfolio ID", required = false) + @RequestParam(value = "portfolioId") String portfolioId, + @Parameter(description = "User ID", required = false) + @RequestParam(value = "userId") String userId) { + return ResponseEntity.ok(documentProcessorService.processBatchDocuments(files, documentType, portfolioId, userId)); } @Operation( diff --git a/src/main/java/org/am/mypotrfolio/domain/common/DocumentRequest.java b/src/main/java/org/am/mypotrfolio/domain/common/DocumentRequest.java index 6a0bc2e..dff7c1b 100644 --- a/src/main/java/org/am/mypotrfolio/domain/common/DocumentRequest.java +++ b/src/main/java/org/am/mypotrfolio/domain/common/DocumentRequest.java @@ -22,4 +22,6 @@ public class DocumentRequest { private BrokerType brokerType; private DocumentType documentType; private MultipartFile file; + private String portfolioId; + private String userId; } diff --git a/src/main/java/org/am/mypotrfolio/domain/common/DocumentType.java b/src/main/java/org/am/mypotrfolio/domain/common/DocumentType.java index 403d641..803e862 100644 --- a/src/main/java/org/am/mypotrfolio/domain/common/DocumentType.java +++ b/src/main/java/org/am/mypotrfolio/domain/common/DocumentType.java @@ -6,7 +6,9 @@ public enum DocumentType { NPS_STATEMENT("NPS_Statement"), COMPANY_FINANCIAL_REPORT("Company_Financial_Report"), STOCK_PORTFOLIO("Stock_Portfolio"), - NSE_INDICES("NSE_Indices"); + NSE_INDICES("NSE_Indices"), + TRADE_FNO("Trade_FNO"), + TRADE_EQ("Trade_EQ"); private String documentType; @@ -51,4 +53,12 @@ public boolean isStockPortfolio() { public boolean isNseIndices() { return "NSE_Indices".equals(documentType); } + + public boolean isTradeFno() { + return "Trade_FNO".equals(documentType); + } + + public boolean isTradeEq() { + return "Trade_EQ".equals(documentType); + } } diff --git a/src/main/java/org/am/mypotrfolio/kafka/config/KafkaConfig.java b/src/main/java/org/am/mypotrfolio/kafka/config/KafkaConfig.java index db9eaa4..d25aa4f 100644 --- a/src/main/java/org/am/mypotrfolio/kafka/config/KafkaConfig.java +++ b/src/main/java/org/am/mypotrfolio/kafka/config/KafkaConfig.java @@ -1,9 +1,9 @@ package org.am.mypotrfolio.kafka.config; +import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.admin.NewTopic; -import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.producer.ProducerConfig; -import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.config.SaslConfigs; import org.apache.kafka.common.serialization.StringSerializer; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; @@ -22,23 +22,52 @@ public class KafkaConfig { @Value("${spring.kafka.bootstrap-servers}") private String bootstrapServers; + @Value("${spring.kafka.properties.security.protocol}") + private String securityProtocol; + + @Value("${spring.kafka.properties.sasl.mechanism}") + private String saslMechanism; + + @Value("${spring.kafka.properties.sasl.jaas.config}") + private String jaasConfig; + @Value("${spring.kafka.consumer.group-id}") private String groupId; - @Value("${app.kafka.topic}") - private String topicName; + @Value("${app.kafka.portfolio-topic}") + private String portfolioTopic; + + @Value("${app.kafka.trade-topic}") + private String tradeTopic; @Bean public NewTopic createTopic() { - return new NewTopic(topicName, 1, (short) 1); + return new NewTopic(portfolioTopic, 1, (short) 1); + } + + @Bean + public NewTopic createTradeTopic() { + return new NewTopic(tradeTopic, 1, (short) 1); + } + + @Bean + public Map kafkaConfigs() { + Map props = new HashMap<>(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); + + if(jaasConfig != null && !jaasConfig.isEmpty()) { + props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol); + props.put(SaslConfigs.SASL_MECHANISM, saslMechanism); + props.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig); + } + return props; } @Bean public ProducerFactory producerFactory() { - Map configProps = new HashMap<>(); - configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); - configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); - configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); + Map configProps = kafkaConfigs(); return new DefaultKafkaProducerFactory<>(configProps); } @@ -49,11 +78,7 @@ public KafkaTemplate kafkaTemplate() { @Bean public ConsumerFactory consumerFactory() { - Map props = new HashMap<>(); - props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); - props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); - props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); - props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class); + Map props = kafkaConfigs(); props.put(JsonDeserializer.TRUSTED_PACKAGES, "*"); return new DefaultKafkaConsumerFactory<>(props); } diff --git a/src/main/java/org/am/mypotrfolio/kafka/model/PortfolioUpdateEvent.java b/src/main/java/org/am/mypotrfolio/kafka/model/PortfolioUpdateEvent.java index 6adf4e6..f853e0b 100644 --- a/src/main/java/org/am/mypotrfolio/kafka/model/PortfolioUpdateEvent.java +++ b/src/main/java/org/am/mypotrfolio/kafka/model/PortfolioUpdateEvent.java @@ -24,6 +24,7 @@ public class PortfolioUpdateEvent { private UUID id; private BrokerType brokerType; private String userId; + private String portfolioId; private List equities; private List mutualFunds; private LocalDateTime timestamp; diff --git a/src/main/java/org/am/mypotrfolio/kafka/model/TradeUpdateEvent.java b/src/main/java/org/am/mypotrfolio/kafka/model/TradeUpdateEvent.java new file mode 100644 index 0000000..1c1804e --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/kafka/model/TradeUpdateEvent.java @@ -0,0 +1,30 @@ +package org.am.mypotrfolio.kafka.model; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.UUID; + +import org.am.mypotrfolio.model.trade.TradeModel; + +import com.am.common.amcommondata.model.enums.BrokerType; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * Event model for trade updates to be sent via Kafka + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class TradeUpdateEvent { + private UUID id; + private String userId; + private BrokerType brokerType; + private String portfolioId; + private LocalDateTime timestamp; + private List trades; +} diff --git a/src/main/java/org/am/mypotrfolio/kafka/producer/KafkaProducerService.java b/src/main/java/org/am/mypotrfolio/kafka/producer/KafkaProducerService.java index 053b7a5..60b7dd6 100644 --- a/src/main/java/org/am/mypotrfolio/kafka/producer/KafkaProducerService.java +++ b/src/main/java/org/am/mypotrfolio/kafka/producer/KafkaProducerService.java @@ -4,10 +4,12 @@ import lombok.extern.slf4j.Slf4j; import org.am.mypotrfolio.kafka.model.PortfolioUpdateEvent; +import org.am.mypotrfolio.kafka.model.TradeUpdateEvent; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.header.internals.RecordHeaders; import org.springframework.beans.factory.annotation.Value; import org.springframework.kafka.core.KafkaTemplate; +import java.util.UUID; import org.springframework.stereotype.Service; @Slf4j @@ -17,19 +19,41 @@ public class KafkaProducerService { private final KafkaTemplate kafkaTemplate; - @Value("${app.kafka.topic}") - private String topicName; + @Value("${app.kafka.portfolio-topic}") + private String portfolioTopic; + + @Value("${app.kafka.trade-topic}") + private String tradeTopic; public void sendMessage(PortfolioUpdateEvent portfolioUpdateEvent) { + RecordHeaders headers = buildCommonHeaders( + portfolioUpdateEvent.getId(), + portfolioUpdateEvent.getUserId(), + portfolioUpdateEvent.getTimestamp() + ); + sendKafkaMessage(portfolioTopic, portfolioUpdateEvent.getId().toString(), portfolioUpdateEvent, headers); + } + + public void sendTradeUpdateEvent(TradeUpdateEvent tradeUpdateEvent) { + RecordHeaders headers = buildCommonHeaders( + tradeUpdateEvent.getId(), + tradeUpdateEvent.getUserId(), + tradeUpdateEvent.getTimestamp() + ); + sendKafkaMessage(tradeTopic, tradeUpdateEvent.getId().toString(), tradeUpdateEvent, headers); + } + + private RecordHeaders buildCommonHeaders(UUID id, String userId, Object timestamp) { RecordHeaders headers = new RecordHeaders(); - headers.add("id", portfolioUpdateEvent.getId().toString().getBytes()); - headers.add("userId", portfolioUpdateEvent.getUserId().getBytes()); - headers.add("timestamp", String.valueOf(portfolioUpdateEvent.getTimestamp()).getBytes()); + headers.add("id", id.toString().getBytes()); + headers.add("userId", userId.getBytes()); + headers.add("timestamp", String.valueOf(timestamp).getBytes()); + return headers; + } - ProducerRecord record = new ProducerRecord<>(topicName, null, - portfolioUpdateEvent.getId().toString(), portfolioUpdateEvent, headers); - + private void sendKafkaMessage(String topicName, String key, Object event, RecordHeaders headers) { + ProducerRecord record = new ProducerRecord<>(topicName, null, key, event, headers); kafkaTemplate.send(record) .whenComplete((result, ex) -> { if (ex == null) { diff --git a/src/main/java/org/am/mypotrfolio/mapper/TradeMapper.java b/src/main/java/org/am/mypotrfolio/mapper/TradeMapper.java new file mode 100644 index 0000000..72aca8c --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/mapper/TradeMapper.java @@ -0,0 +1,225 @@ +package org.am.mypotrfolio.mapper; + +import org.am.mypotrfolio.model.trade.*; +import org.springframework.stereotype.Component; + +import com.am.common.amcommondata.model.enums.BrokerType; + +import java.math.BigDecimal; +import java.time.LocalDate; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeParseException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Mapper utility class for converting between Trade and TradeModel objects. + */ +@Component +public class TradeMapper { + + /** + * Converts a Trade object to a TradeModel object. + * + * @param trade the Trade object to convert + * @return the converted TradeModel object + */ + public TradeModel toTradeModel(Trade trade, BrokerType brokerType) { + if (trade == null) { + return null; + } + + return TradeModel.builder() + .basicInfo(buildBasicInfo(trade, brokerType)) + .instrumentInfo(buildInstrumentInfo(trade)) + .executionInfo(buildExecutionInfo(trade)) + .build(); + } + + private TradeModel.BasicInfo buildBasicInfo(Trade trade, BrokerType brokerType) { + return TradeModel.BasicInfo.builder() + .tradeId(trade.getTradeId()) + .orderId(trade.getOrderId()) + .tradeDate(trade.getTradeDate()) + .orderExecutionTime(trade.getOrderExecutionTime()) + .brokerType(brokerType) // Set appropriate broker type if available + .tradeType(TradeType.valueOf(trade.getTradeType().toUpperCase())) // Set appropriate trade type if available + .build(); + } + + + private TradeModel.InstrumentInfo buildInstrumentInfo(Trade trade) { + TradeModel.InstrumentInfo.InstrumentInfoBuilder builder = TradeModel.InstrumentInfo.builder() + .symbol(trade.getSymbol()) + .isin(trade.getIsin()) + .exchange(trade.getExchange()) + .segment(trade.getSegment()) + .series(trade.getSeries()); + + // If segment is F&O, add FnO info + if (trade.getSegment() == Segment.FUTURES || + trade.getSegment() == Segment.OPTIONS || + trade.getSegment() == Segment.FNO || + "FO".equalsIgnoreCase(trade.getSegment().getValue())) { + builder.fnoInfo(buildFnOInfo(trade)); + } + + return builder.build(); + } + + private TradeModel.ExecutionInfo buildExecutionInfo(Trade trade) { + String symbol = trade.getSymbol(); + BigDecimal lotSize = determineLotSize(symbol, trade.getTradeDate()); + TradeModel.ExecutionInfo.ExecutionInfoBuilder executionInfoBuilder = TradeModel.ExecutionInfo.builder() + .tradeType(TradeType.valueOf(trade.getTradeType().toUpperCase())) + .auction(trade.getAuction()) + .quantity(trade.getQuantity().intValue()) + .price(trade.getPrice()); + + if (lotSize != null) { + executionInfoBuilder.lotSize(trade.getQuantity().intValue() / lotSize.intValue()); + } + + return executionInfoBuilder.build(); + } + + /** + * Builds FnO information by parsing the trade symbol. + * Examples: + * - RELIANCE20AUGFUT -> Equity Future + * - BANKNIFTY20AUG23000PE -> Index Option + * + * @param trade the trade object + * @return FnOInfo object with parsed details + */ + private TradeModel.FnOInfo buildFnOInfo(Trade trade) { + String symbol = trade.getSymbol(); + if (symbol == null || symbol.isEmpty()) { + return null; + } + + TradeModel.FnOInfo.FnOInfoBuilder builder = TradeModel.FnOInfo.builder(); + + // Check if it's a future (ends with FUT) + if (symbol.endsWith("FUT")) { + // It's a future + String baseSymbol = extractBaseSymbol(symbol, "FUT"); + LocalDate expiryDate = extractExpiryDate(symbol); + + // Determine if it's an index future or equity future + FNOTradeType instrumentType = isIndex(baseSymbol) ? FNOTradeType.FUTIDX : FNOTradeType.FUTEQ; + + builder.instrumentType(instrumentType) + .expiryDate(expiryDate) + .optionType(OptionType.NONE); + + } else if (symbol.endsWith("CE") || symbol.endsWith("PE")) { + // It's an option + OptionType optionType = symbol.endsWith("CE") ? OptionType.CALL : OptionType.PUT; + String baseSymbol; + BigDecimal strikePrice = null; + + // Extract strike price - it's the numeric part before CE/PE + Pattern pattern = Pattern.compile("(\\d+)(CE|PE)$"); + Matcher matcher = pattern.matcher(symbol); + if (matcher.find()) { + strikePrice = new BigDecimal(matcher.group(1)); + baseSymbol = symbol.substring(0, symbol.length() - matcher.group().length()); + } else { + baseSymbol = extractBaseSymbol(symbol, optionType.getValue()); + } + + LocalDate expiryDate = extractExpiryDate(symbol); + + // Determine if it's an index option or equity option + FNOTradeType instrumentType = isIndex(baseSymbol) ? FNOTradeType.OPTIDX : FNOTradeType.OPTEQ; + + builder.instrumentType(instrumentType) + .expiryDate(expiryDate) + .strikePrice(strikePrice) + .optionType(optionType); + } + + // Set lot size based on trade date + builder.lotSize(determineLotSize(symbol, trade.getTradeDate())); + + return builder.build(); + } + + /** + * Extracts the base symbol from the F&O symbol + */ + private String extractBaseSymbol(String symbol, String suffix) { + // Remove the suffix and any date/month information + String baseSymbol = symbol.replace(suffix, ""); + + // Find where the date/month part starts (usually after letters) + Pattern pattern = Pattern.compile("^([A-Za-z&]+)"); + Matcher matcher = pattern.matcher(baseSymbol); + if (matcher.find()) { + return matcher.group(1); + } + + return baseSymbol; + } + + /** + * Extracts the expiry date from the symbol + */ + private LocalDate extractExpiryDate(String symbol) { + // This is a simplified implementation - actual implementation would need to handle + // various date formats like 20AUG, 20AUG23, etc. + try { + // Try to find a date pattern like 20AUG or 20AUG23 + Pattern pattern = Pattern.compile("(\\d{2})([A-Za-z]{3})(\\d{0,2})"); + Matcher matcher = pattern.matcher(symbol); + if (matcher.find()) { + String day = matcher.group(1); + String month = matcher.group(2); + String year = matcher.group(3); + + if (year.isEmpty()) { + // If year is not specified, use current year + year = String.valueOf(LocalDate.now().getYear() % 100); + } + + // Parse the date + String dateStr = day + "-" + month + "-" + "20" + year; + return LocalDate.parse(dateStr, DateTimeFormatter.ofPattern("dd-MMM-yy")); + } + } catch (DateTimeParseException e) { + // Log error and return null or today's date + } + + return null; + } + + /** + * Determines if the symbol is an index + */ + private boolean isIndex(String symbol) { + return IndexType.isIndex(symbol); + } + + /** + * Determines the lot size based on the symbol and trade date + * + * @param symbol the trade symbol + * @param tradeDate the date of the trade + * @return the lot size applicable for the symbol on the given date + */ + private BigDecimal determineLotSize(String symbol, LocalDate tradeDate) { + String baseSymbol = extractBaseSymbol(symbol, ""); + return IndexType.getLotSizeForSymbol(baseSymbol, tradeDate); + } + + /** + * Determines the current lot size based on the symbol + * + * @param symbol the trade symbol + * @return the current lot size applicable for the symbol + */ + private BigDecimal determineLotSize(String symbol) { + return determineLotSize(symbol, LocalDate.now()); + } +} diff --git a/src/main/java/org/am/mypotrfolio/model/trade/FNOTradeType.java b/src/main/java/org/am/mypotrfolio/model/trade/FNOTradeType.java new file mode 100644 index 0000000..bf312a9 --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/model/trade/FNOTradeType.java @@ -0,0 +1,46 @@ +package org.am.mypotrfolio.model.trade; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; + +import java.util.Arrays; + +/** + * Enum representing the type of F&O instrument + */ +public enum FNOTradeType { + FUTIDX("FUTIDX", "Index Futures"), + OPTIDX("OPTIDX", "Index Options"), + FUTEQ("FUTEQ", "Equity Futures"), + OPTEQ("OPTEQ", "Equity Options"), + UNKNOWN("UNKNOWN", "Unknown Instrument Type"); + + private final String value; + private final String description; + + FNOTradeType(String value, String description) { + this.value = value; + this.description = description; + } + + @JsonValue + public String getValue() { + return value; + } + + public String getDescription() { + return description; + } + + @JsonCreator + public static FNOTradeType fromValue(String value) { + if (value == null) { + return UNKNOWN; + } + + return Arrays.stream(values()) + .filter(type -> type.value.equalsIgnoreCase(value)) + .findFirst() + .orElse(UNKNOWN); + } +} diff --git a/src/main/java/org/am/mypotrfolio/model/trade/IndexType.java b/src/main/java/org/am/mypotrfolio/model/trade/IndexType.java new file mode 100644 index 0000000..d19bb13 --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/model/trade/IndexType.java @@ -0,0 +1,139 @@ +package org.am.mypotrfolio.model.trade; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; + +import java.math.BigDecimal; +import java.time.LocalDate; +import java.util.Arrays; + +import org.am.mypotrfolio.config.LotSizeConfig; +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.lang.NonNull; + +/** + * Enum representing different index types in the market + */ +public enum IndexType implements ApplicationContextAware { + NIFTY("NIFTY", "Nifty 50"), + BANKNIFTY("BANKNIFTY", "Bank Nifty"), + FINNIFTY("FINNIFTY", "Financial Services Nifty"), + MIDCPNIFTY("MIDCPNIFTY", "Midcap Nifty"), + UNKNOWN("UNKNOWN", "Unknown Index"); + + private static LotSizeConfig lotSizeConfig; + + private final String value; + private final String description; + + IndexType(String value, String description) { + this.value = value; + this.description = description; + } + + @Override + public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException { + // This method will be called once for any enum constant + // We only need to set the lotSizeConfig once + if (IndexType.lotSizeConfig == null) { + IndexType.lotSizeConfig = applicationContext.getBean(LotSizeConfig.class); + } + } + + @JsonValue + public String getValue() { + return value; + } + + public String getDescription() { + return description; + } + + /** + * Get the lot size for this index on the given date + * + * @param date the date for which to get the lot size + * @return the lot size applicable on the given date + */ + public BigDecimal getLotSize(LocalDate date) { + if (lotSizeConfig == null) { + // Fallback to default values if config is not available + switch (this) { + case NIFTY: return new BigDecimal(50); + case BANKNIFTY: return new BigDecimal(25); + case FINNIFTY: return new BigDecimal(40); + case MIDCPNIFTY: return new BigDecimal(75); + default: return BigDecimal.ONE; + } + } + return lotSizeConfig.getLotSize(this.value, date); + } + + /** + * Get the current lot size for this index + * + * @return the current lot size + */ + public BigDecimal getLotSize() { + return getLotSize(LocalDate.now()); + } + + @JsonCreator + public static IndexType fromValue(String value) { + if (value == null) { + return UNKNOWN; + } + + return Arrays.stream(values()) + .filter(indexType -> indexType.value.equalsIgnoreCase(value)) + .findFirst() + .orElse(UNKNOWN); + } + + /** + * Check if a symbol represents an index + * + * @param symbol the symbol to check + * @return true if the symbol is an index, false otherwise + */ + public static boolean isIndex(String symbol) { + if (symbol == null) { + return false; + } + + return Arrays.stream(values()) + .filter(indexType -> !indexType.equals(UNKNOWN)) + .anyMatch(indexType -> indexType.value.equalsIgnoreCase(symbol)); + } + + /** + * Get the lot size for a given symbol on the given date + * + * @param symbol the symbol to get the lot size for + * @param date the date for which to get the lot size + * @return the lot size for the symbol on the given date, or a default value if not found + */ + public static BigDecimal getLotSizeForSymbol(String symbol, LocalDate date) { + if (symbol == null) { + return BigDecimal.ONE; + } + + return Arrays.stream(values()) + .filter(indexType -> indexType.value.equalsIgnoreCase(symbol)) + .findFirst() + .map(indexType -> indexType.getLotSize(date)) + .orElse(null); // Default lot size for equity + } + + /** + * Get the current lot size for a given symbol + * + * @param symbol the symbol to get the lot size for + * @return the current lot size for the symbol, or a default value if not found + */ + public static BigDecimal getLotSizeForSymbol(String symbol) { + return getLotSizeForSymbol(symbol, LocalDate.now()); + } +} diff --git a/src/main/java/org/am/mypotrfolio/model/trade/OptionType.java b/src/main/java/org/am/mypotrfolio/model/trade/OptionType.java new file mode 100644 index 0000000..96d02ba --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/model/trade/OptionType.java @@ -0,0 +1,46 @@ +package org.am.mypotrfolio.model.trade; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; + +import java.util.Arrays; + +public enum OptionType { + CALL("CE", "Call Option"), + PUT("PE", "Put Option"), + NONE(null, "Not Applicable"); + + private final String value; + private final String description; + + OptionType(String value, String description) { + this.value = value; + this.description = description; + } + + @JsonValue + public String getValue() { + return value; + } + + public String getDescription() { + return description; + } + + @JsonCreator + public static OptionType fromValue(String value) { + if (value == null) { + return NONE; + } + + return Arrays.stream(values()) + .filter(optionType -> { + if (optionType.value == null) { + return false; + } + return optionType.value.equalsIgnoreCase(value); + }) + .findFirst() + .orElse(NONE); + } +} diff --git a/src/main/java/org/am/mypotrfolio/model/trade/Segment.java b/src/main/java/org/am/mypotrfolio/model/trade/Segment.java new file mode 100644 index 0000000..013c516 --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/model/trade/Segment.java @@ -0,0 +1,39 @@ +package org.am.mypotrfolio.model.trade; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; + +import java.util.Arrays; + +public enum Segment { + EQUITY("EQUITY"), + FUTURES("FUTURES"), + OPTIONS("OPTIONS"), + CURRENCY("CURRENCY"), + COMMODITY("COMMODITY"), + FNO("FO"), + UNKNOWN("UNKNOWN"); + + private final String value; + + Segment(String value) { + this.value = value; + } + + @JsonValue + public String getValue() { + return value; + } + + @JsonCreator + public static Segment fromValue(String value) { + if (value == null) { + return UNKNOWN; + } + + return Arrays.stream(values()) + .filter(segment -> segment.value.equalsIgnoreCase(value)) + .findFirst() + .orElse(UNKNOWN); + } +} diff --git a/src/main/java/org/am/mypotrfolio/model/trade/Series.java b/src/main/java/org/am/mypotrfolio/model/trade/Series.java new file mode 100644 index 0000000..00ce752 --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/model/trade/Series.java @@ -0,0 +1,49 @@ +package org.am.mypotrfolio.model.trade; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; + +import java.util.Arrays; + +public enum Series { + EQ("EQ", "Equity"), + BE("BE", "Book Entry"), + BL("BL", "Block Deal"), + BO("BO", "Buyout"), + BT("BT", "Bond Trading"), + GC("GC", "Government Securities"), + IL("IL", "Index Linked"), + IQ("IQ", "Interest Quote"), + IT("IT", "Index Trading"), + SM("SM", "SLB Market"), + UNKNOWN("UNKNOWN", "Unknown Series"); + + private final String value; + private final String description; + + Series(String value, String description) { + this.value = value; + this.description = description; + } + + @JsonValue + public String getValue() { + return value; + } + + public String getDescription() { + return description; + } + + @JsonCreator + public static Series fromValue(String value) { + if (value == null) { + return UNKNOWN; + } + + return Arrays.stream(values()) + .filter(series -> series.value.equalsIgnoreCase(value)) + .findFirst() + .orElse(UNKNOWN); + } +} diff --git a/src/main/java/org/am/mypotrfolio/model/trade/Trade.java b/src/main/java/org/am/mypotrfolio/model/trade/Trade.java new file mode 100644 index 0000000..9bf8b85 --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/model/trade/Trade.java @@ -0,0 +1,71 @@ +package org.am.mypotrfolio.model.trade; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.math.BigDecimal; +import java.time.LocalDate; +import java.time.LocalDateTime; + +import com.fasterxml.jackson.annotation.JsonAlias; +import com.fasterxml.jackson.annotation.JsonProperty; + +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class Trade { + @JsonProperty("symbol") + @JsonAlias({"Symbol", "symbol"}) + private String symbol; + + @JsonProperty("isin") + @JsonAlias({"ISIN", "isin"}) + private String isin; + + @JsonProperty("tradeDate") + @JsonAlias({"Trade Date", "tradeDate"}) + private LocalDate tradeDate; + + @JsonProperty("exchange") + @JsonAlias({"Exchange", "exchange"}) + private String exchange; + + @JsonProperty("segment") + @JsonAlias({"Segment", "segment"}) + private Segment segment; + + @JsonProperty("series") + @JsonAlias({"Series", "series"}) + private Series series; + + @JsonProperty("tradeType") + @JsonAlias({"Trade Type", "tradeType"}) + private String tradeType; + + @JsonProperty("auction") + @JsonAlias({"Auction", "auction"}) + private String auction; + + @JsonProperty("quantity") + @JsonAlias({"Quantity", "quantity"}) + private BigDecimal quantity; + + @JsonProperty("price") + @JsonAlias({"Price", "price"}) + private BigDecimal price; + + @JsonProperty("tradeId") + @JsonAlias({"Trade Id", "tradeId", "Trade ID"}) + private String tradeId; + + @JsonProperty("orderId") + @JsonAlias({"Order Id", "orderId","Order ID"}) + private String orderId; + + @JsonProperty("orderExecutionTime") + @JsonAlias({"Order Execution Time", "orderExecutionTime", "Order Execution Time"}) + private LocalDateTime orderExecutionTime; +} \ No newline at end of file diff --git a/src/main/java/org/am/mypotrfolio/model/trade/TradeModel.java b/src/main/java/org/am/mypotrfolio/model/trade/TradeModel.java new file mode 100644 index 0000000..5938440 --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/model/trade/TradeModel.java @@ -0,0 +1,102 @@ +package org.am.mypotrfolio.model.trade; + +import java.math.BigDecimal; +import java.time.LocalDate; +import java.time.LocalDateTime; + +import com.am.common.amcommondata.model.enums.BrokerType; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * Model class representing a trade based on Zerodha's F&O trade book structure + */ +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class TradeModel { + private BasicInfo basicInfo; + private InstrumentInfo instrumentInfo; + private ExecutionInfo executionInfo; + private Charges charges; + private Financials financials; + + @Data + @Builder + @NoArgsConstructor + @AllArgsConstructor + public static class BasicInfo { + private String tradeId; + private String orderId; + private LocalDate tradeDate; + private LocalDateTime orderExecutionTime; + private BrokerType brokerType; + private TradeType tradeType; + } + + @Data + @Builder + @NoArgsConstructor + @AllArgsConstructor + public static class InstrumentInfo { + private String symbol; + private String isin; + private String exchange; + private Segment segment; + private Series series; + private FnOInfo fnoInfo; + + } + + @Data + @Builder + @NoArgsConstructor + @AllArgsConstructor + public static class ExecutionInfo { + private TradeType tradeType; + private String auction; + private Integer quantity; + private BigDecimal price; + private Integer lotSize; + } + + @Data + @Builder + @NoArgsConstructor + @AllArgsConstructor + public static class FnOInfo { + private FNOTradeType instrumentType; // FUTIDX, OPTIDX, FUTEQ, OPTEQ + private LocalDate expiryDate; + private BigDecimal strikePrice; + private OptionType optionType; // CALL, PUT, NONE for futures + private BigDecimal lotSize; + private BigDecimal premiumValue; + } + + @Data + @Builder + @NoArgsConstructor + @AllArgsConstructor + public static class Charges { + private BigDecimal brokerage; + private BigDecimal stt; + private BigDecimal transactionCharges; + private BigDecimal stampDuty; + private BigDecimal sebiCharges; + private BigDecimal gst; + private BigDecimal totalTaxes; + } + + @Data + @Builder + @NoArgsConstructor + @AllArgsConstructor + public static class Financials { + private BigDecimal turnover; + private BigDecimal netAmount; + } +} diff --git a/src/main/java/org/am/mypotrfolio/model/trade/TradeType.java b/src/main/java/org/am/mypotrfolio/model/trade/TradeType.java new file mode 100644 index 0000000..24f9fea --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/model/trade/TradeType.java @@ -0,0 +1,15 @@ +package org.am.mypotrfolio.model.trade; + +/** + * Enum representing the type of trade (buy or sell) + */ +public enum TradeType { + BUY("BUY"), + SELL("SELL"); + + private final String value; + + TradeType(String value) { + this.value = value; + } +} diff --git a/src/main/java/org/am/mypotrfolio/nsesecurity/domain/NseSecurity.java b/src/main/java/org/am/mypotrfolio/nsesecurity/domain/NseSecurity.java deleted file mode 100644 index 7051786..0000000 --- a/src/main/java/org/am/mypotrfolio/nsesecurity/domain/NseSecurity.java +++ /dev/null @@ -1,28 +0,0 @@ -package org.am.mypotrfolio.nsesecurity.domain; - -import lombok.Data; -import lombok.Builder; -import lombok.NoArgsConstructor; -import lombok.AllArgsConstructor; - -@Data -@Builder -@NoArgsConstructor -@AllArgsConstructor -public class NseSecurity { - private String securityId; - private String securityName; - private String status; - private String series; - private String isin; - private Double faceValue; - private String industry; - private String instrumentType; - private String sectorName; - private String industryNewName; - private String industryGroupName; - private String industrySubGroupName; - private String sectorIndices; - private String thematicIndices; - private String marketIndices; -} \ No newline at end of file diff --git a/src/main/java/org/am/mypotrfolio/nsesecurity/entity/EquityDataEntity.java b/src/main/java/org/am/mypotrfolio/nsesecurity/entity/EquityDataEntity.java deleted file mode 100644 index 7479032..0000000 --- a/src/main/java/org/am/mypotrfolio/nsesecurity/entity/EquityDataEntity.java +++ /dev/null @@ -1,89 +0,0 @@ -package org.am.mypotrfolio.nsesecurity.entity; - -import jakarta.persistence.*; -import lombok.Getter; -import lombok.Setter; - -import java.time.ZonedDateTime; -import java.util.UUID; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.am.common.amcommondata.model.MarketCapType; - -@Entity -@Table(name = "equity_data") -@Getter -@Setter -public class EquityDataEntity { - - @Id - @GeneratedValue(strategy = GenerationType.AUTO) - private UUID id; - - @Column(nullable = false) - private String symbol; - - @Column(unique = true) - private String isin; - - @Column(nullable = false) - private String name; - - private String series; - - private Double faceValue; - - private Double marketCap; - - private String industry; - - private String sector; - - @Column(name = "instrument_type") - private String instrumentType; - - @Enumerated(EnumType.STRING) - private MarketCapType marketCapType; - - @Column(name = "created_at") - private ZonedDateTime createdAt; - - @Column(name = "updated_at") - private ZonedDateTime updatedAt; - - private static final Logger log = LoggerFactory.getLogger(EquityDataEntity.class); - - @PrePersist - protected void onCreate() { - createdAt = ZonedDateTime.now(); - updatedAt = ZonedDateTime.now(); - updateMarketCapType(); - } - - @PreUpdate - protected void onUpdate() { - updatedAt = ZonedDateTime.now(); - updateMarketCapType(); - } - - public void updateMarketCapType() { - if (marketCap != null && marketCap > 0) { - // Calculate market cap type based on market cap value - if (marketCap >= 1000000000000D) { // > 1 lakh crore - marketCapType = MarketCapType.LARGE_CAP; - } else if (marketCap >= 250000000000D) { // > 25k crore - marketCapType = MarketCapType.MID_CAP; - } else if (marketCap >= 50000000000D) { // > 5k crore - marketCapType = MarketCapType.SMALL_CAP; - } else { - marketCapType = MarketCapType.MICRO_CAP; - } - log.info("Setting market cap type for {}: {} (Market Cap: {})", symbol, marketCapType, marketCap); - } else { - marketCapType = MarketCapType.MICRO_CAP; - log.info("No market cap available for {}, defaulting to MICRO_CAP", symbol); - } - } -} \ No newline at end of file diff --git a/src/main/java/org/am/mypotrfolio/nsesecurity/entity/NseSecurityEntity.java b/src/main/java/org/am/mypotrfolio/nsesecurity/entity/NseSecurityEntity.java deleted file mode 100644 index 370cd62..0000000 --- a/src/main/java/org/am/mypotrfolio/nsesecurity/entity/NseSecurityEntity.java +++ /dev/null @@ -1,52 +0,0 @@ -package org.am.mypotrfolio.nsesecurity.entity; - -import jakarta.persistence.*; -import lombok.Data; -import lombok.Getter; - -import java.time.ZonedDateTime; -import java.util.UUID; - -@Data -@Entity -@Table(name = "nse_security_data") -@Getter -public class NseSecurityEntity { - - @Id - @GeneratedValue(strategy = GenerationType.AUTO) - private UUID id; - - private String securityId; - private String securityName; - private String status; - private String series; - private String isin; - private Double faceValue; - private String industry; - private String instrumentType; - private String sectorName; - private String industryNewName; - private String iGroupName; - private String iSubGroupName; - private String sectorIndices; - private String thematicIndices; - private String marketIndices; - - @Column(name = "created_at") - private ZonedDateTime createdAt; - - @Column(name = "updated_at") - private ZonedDateTime updatedAt; - - @PrePersist - protected void onCreate() { - createdAt = ZonedDateTime.now(); - updatedAt = createdAt; - } - - @PreUpdate - protected void onUpdate() { - updatedAt = ZonedDateTime.now(); - } -} \ No newline at end of file diff --git a/src/main/java/org/am/mypotrfolio/nsesecurity/repo/EquityDataRepository.java b/src/main/java/org/am/mypotrfolio/nsesecurity/repo/EquityDataRepository.java deleted file mode 100644 index 60bfff0..0000000 --- a/src/main/java/org/am/mypotrfolio/nsesecurity/repo/EquityDataRepository.java +++ /dev/null @@ -1,19 +0,0 @@ -package org.am.mypotrfolio.nsesecurity.repo; - -import org.am.mypotrfolio.nsesecurity.entity.EquityDataEntity; -import org.springframework.data.jpa.repository.JpaRepository; -import org.springframework.data.jpa.repository.Query; -import org.springframework.data.repository.query.Param; -import org.springframework.stereotype.Repository; - -import java.util.Optional; -import java.util.UUID; - -@Repository -public interface EquityDataRepository extends JpaRepository { - Optional findByIsin(String isin); - Optional findBySymbol(String symbol); - - @Query("SELECT e FROM EquityDataEntity e WHERE e.isin = :key OR e.symbol = :key") - Optional findByKey(@Param("key") String key); -} \ No newline at end of file diff --git a/src/main/java/org/am/mypotrfolio/nsesecurity/repo/NseSecurityRepository.java b/src/main/java/org/am/mypotrfolio/nsesecurity/repo/NseSecurityRepository.java deleted file mode 100644 index e0ad95b..0000000 --- a/src/main/java/org/am/mypotrfolio/nsesecurity/repo/NseSecurityRepository.java +++ /dev/null @@ -1,89 +0,0 @@ -package org.am.mypotrfolio.nsesecurity.repo; - -import org.am.mypotrfolio.nsesecurity.domain.NseSecurity; -import org.am.mypotrfolio.nsesecurity.entity.NseSecurityEntity; -import org.springframework.data.jpa.repository.JpaRepository; -import org.springframework.data.jpa.repository.Query; -import org.springframework.data.repository.query.Param; -import org.springframework.stereotype.Repository; - -import java.util.List; -import java.util.Optional; -import java.util.UUID; - -@Repository -public interface NseSecurityRepository extends JpaRepository { - - @Query("SELECT new org.am.mypotrfolio.nsesecurity.domain.NseSecurity(" + - "n.securityId, n.securityName, n.status, n.series, n.isin, n.faceValue, " + - "n.industry, n.instrumentType, n.sectorName, n.industryNewName, n.iGroupName, " + - "n.iSubGroupName, n.sectorIndices, n.thematicIndices, n.marketIndices) " + - "FROM NseSecurityEntity n WHERE n.isin = :isin AND n.status = 'Active'") - Optional findByIsin(@Param("isin") String isin); - - @Query("SELECT new org.am.mypotrfolio.nsesecurity.domain.NseSecurity(" + - "n.securityId, n.securityName, n.status, n.series, n.isin, n.faceValue, " + - "n.industry, n.instrumentType, n.sectorName, n.industryNewName, n.iGroupName, " + - "n.iSubGroupName, n.sectorIndices, n.thematicIndices, n.marketIndices) " + - "FROM NseSecurityEntity n WHERE LOWER(n.securityName) = LOWER(:securityName) AND n.status = 'Active'") - Optional findBySecurityName(@Param("securityName") String securityName); - - @Query("SELECT new org.am.mypotrfolio.nsesecurity.domain.NseSecurity(" + - "n.securityId, n.securityName, n.status, n.series, n.isin, n.faceValue, " + - "n.industry, n.instrumentType, n.sectorName, n.industryNewName, n.iGroupName, " + - "n.iSubGroupName, n.sectorIndices, n.thematicIndices, n.marketIndices) " + - "FROM NseSecurityEntity n WHERE LOWER(n.securityName) LIKE LOWER(CONCAT('%', :partialName, '%')) AND n.status = 'Active' ORDER BY LENGTH(n.securityName)") - List findBySecurityNameFuzzy(@Param("partialName") String partialName); - - @Query("SELECT new org.am.mypotrfolio.nsesecurity.domain.NseSecurity(" + - "n.securityId, n.securityName, n.status, n.series, n.isin, n.faceValue, " + - "n.industry, n.instrumentType, n.sectorName, n.industryNewName, n.iGroupName, " + - "n.iSubGroupName, n.sectorIndices, n.thematicIndices, n.marketIndices) " + - "FROM NseSecurityEntity n WHERE n.securityId = :securityId AND n.status = 'Active'") - Optional findBySecurityId(@Param("securityId") String securityId); - - @Query("SELECT new org.am.mypotrfolio.nsesecurity.domain.NseSecurity(" + - "n.securityId, n.securityName, n.status, n.series, n.isin, n.faceValue, " + - "n.industry, n.instrumentType, n.sectorName, n.industryNewName, n.iGroupName, " + - "n.iSubGroupName, n.sectorIndices, n.thematicIndices, n.marketIndices) " + - "FROM NseSecurityEntity n WHERE n.marketIndices LIKE CONCAT('%', :indexType, '%')") - List findByMarketIndices(@Param("indexType") String marketIndices); - - @Query("SELECT new org.am.mypotrfolio.nsesecurity.domain.NseSecurity(" + - "n.securityId, n.securityName, n.status, n.series, n.isin, n.faceValue, " + - "n.industry, n.instrumentType, n.sectorName, n.industryNewName, n.iGroupName, " + - "n.iSubGroupName, n.sectorIndices, n.thematicIndices, n.marketIndices) " + - "FROM NseSecurityEntity n WHERE n.status = 'Active'") - List findAllActiveSecurities(); - - @Query(""" - SELECT new org.am.mypotrfolio.nsesecurity.domain.NseSecurity( - n.securityId, n.securityName, n.status, n.series, n.isin, n.faceValue, - n.industry, n.instrumentType, n.sectorName, n.industryNewName, n.iGroupName, - n.iSubGroupName, n.sectorIndices, n.thematicIndices, n.marketIndices) - FROM NseSecurityEntity n - WHERE ( - :searchParam = n.isin OR - :searchParam = n.securityId OR - LOWER(n.securityName) LIKE LOWER(CONCAT('%', :searchParam, '%')) - ) - AND n.status = 'Active' - ORDER BY - CASE - WHEN n.isin = :searchParam THEN 1 - WHEN n.securityId = :searchParam THEN 2 - WHEN LOWER(n.securityName) = LOWER(:searchParam) THEN 3 - ELSE 4 - END, - LENGTH(n.securityName) - """) - List findSecurityBySearchParam(@Param("searchParam") String searchParam); - - default Optional findBestMatchBySearchParam(String searchParam) { - if (searchParam == null || searchParam.trim().isEmpty()) { - return Optional.empty(); - } - List matches = findSecurityBySearchParam(searchParam.trim()); - return matches.isEmpty() ? Optional.empty() : Optional.of(matches.get(0)); - } -} \ No newline at end of file diff --git a/src/main/java/org/am/mypotrfolio/processor/AbstractFileProcessor.java b/src/main/java/org/am/mypotrfolio/processor/AbstractFileProcessor.java index 4df7cd4..271fae5 100644 --- a/src/main/java/org/am/mypotrfolio/processor/AbstractFileProcessor.java +++ b/src/main/java/org/am/mypotrfolio/processor/AbstractFileProcessor.java @@ -1,6 +1,9 @@ package org.am.mypotrfolio.processor; import lombok.extern.slf4j.Slf4j; + +import org.am.mypotrfolio.domain.common.DocumentRequest; +import org.am.mypotrfolio.domain.common.DocumentType; import org.springframework.web.multipart.MultipartFile; import com.am.common.amcommondata.model.enums.BrokerType; @@ -11,7 +14,8 @@ public abstract class AbstractFileProcessor implements FileProcessor { @Override - public List> processFile(MultipartFile file, BrokerType brokerType) { + public List> processFile(MultipartFile file, DocumentRequest documentRequest) { + BrokerType brokerType = documentRequest.getBrokerType(); log.info("Processing {} file for broker: {}", getFileType(), brokerType); try { if (brokerType == null) { @@ -20,7 +24,7 @@ public List> processFile(MultipartFile file, BrokerType brok } if (brokerType.isZerodha()) { log.debug("Using Zerodha parser"); - return parseZerodhaFile(file); + return (documentRequest.getDocumentType() == DocumentType.TRADE_EQ || documentRequest.getDocumentType() == DocumentType.TRADE_FNO) ? parseZerodhaTradeFile(file) : parseZerodhaFile(file); } else if (brokerType.isMStock()) { log.debug("Using MStock parser"); return parseMStockFile(file); @@ -40,6 +44,7 @@ public List> processFile(MultipartFile file, BrokerType brok } protected abstract List> parseZerodhaFile(MultipartFile file) throws Exception; + protected abstract List> parseZerodhaTradeFile(MultipartFile file) throws Exception; protected abstract List> parseMStockFile(MultipartFile file) throws Exception; protected abstract List> parseDhanFile(MultipartFile file) throws Exception; protected abstract List> parseGrowFile(MultipartFile file) throws Exception; diff --git a/src/main/java/org/am/mypotrfolio/processor/CsvFileProcessor.java b/src/main/java/org/am/mypotrfolio/processor/CsvFileProcessor.java index fae6860..2f63ddc 100644 --- a/src/main/java/org/am/mypotrfolio/processor/CsvFileProcessor.java +++ b/src/main/java/org/am/mypotrfolio/processor/CsvFileProcessor.java @@ -49,6 +49,10 @@ protected List> parseGrowFile(MultipartFile file) throws Exc protected List> parseNseSecurityFile(MultipartFile file) throws Exception { return parseCsvFile(file, 0); } + @Override + protected List> parseZerodhaTradeFile(MultipartFile file) throws Exception { + return parseCsvFile(file, 15); + } private List> parseCsvFile(MultipartFile file, int skipLines) throws Exception { List> data = new ArrayList<>(); diff --git a/src/main/java/org/am/mypotrfolio/processor/ExcelFileProcessor.java b/src/main/java/org/am/mypotrfolio/processor/ExcelFileProcessor.java index 0c50c79..8305a6d 100644 --- a/src/main/java/org/am/mypotrfolio/processor/ExcelFileProcessor.java +++ b/src/main/java/org/am/mypotrfolio/processor/ExcelFileProcessor.java @@ -3,8 +3,8 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.StringUtils; import org.apache.poi.ss.usermodel.*; +import java.util.stream.Collectors; import org.apache.poi.xssf.usermodel.XSSFWorkbook; import org.springframework.stereotype.Component; import org.springframework.web.multipart.MultipartFile; @@ -55,6 +55,11 @@ protected List> parseNseSecurityFile(MultipartFile file) thr return parseExcelFile(file, 0, 0, 0); } + @Override + protected List> parseZerodhaTradeFile(MultipartFile file) throws Exception { + return parseExcelFile(file, 14, 14, 0); + } + private List> parseExcelFile(MultipartFile file, int headerRow, int skipRows, int skipColumns) throws Exception { List> jsonList = new ArrayList<>(); @@ -76,23 +81,42 @@ private List> parseExcelFile(MultipartFile file, int headerR for (Cell cell : row) { cell.setCellType(CellType.STRING); String header = cell.getStringCellValue().trim(); + // Skip empty header cells + if (header.isEmpty()) { + continue; + } // Remove BOM if present if (headers.isEmpty() && header.startsWith("\uFEFF")) { header = header.substring(1); } headers.add(header); } + // Adjust headers list if we need to skip columns + if (skipColumns > 0) { + headers = headers.subList(skipColumns, headers.size()); + } + // Filter out any remaining empty headers after processing + headers = headers.stream() + .filter(h -> h != null && !h.trim().isEmpty()) + .collect(Collectors.toList()); continue; } if (headers.isEmpty()) continue; - String[] values = new String[headers.size()-skipColumns]; + String[] values = new String[headers.size()]; + int valueIndex = 0; + int cellsToSkip = skipColumns; // Skip first cell + skipColumns + int cellIndex = 0; + for (Cell cell : row) { cell.setCellType(CellType.STRING); - if (cell.getColumnIndex() < headers.size()) { - values[cell.getColumnIndex()-skipColumns] = cell.getStringCellValue(); + // Skip first cell and skipColumns + if (cellIndex >= cellsToSkip && valueIndex < values.length) { + values[valueIndex] = cell.getStringCellValue(); + valueIndex++; } + cellIndex++; } Map rowData = createRowData(headers.toArray(new String[0]), values); diff --git a/src/main/java/org/am/mypotrfolio/processor/FileProcessor.java b/src/main/java/org/am/mypotrfolio/processor/FileProcessor.java index 2db8f75..36340f2 100644 --- a/src/main/java/org/am/mypotrfolio/processor/FileProcessor.java +++ b/src/main/java/org/am/mypotrfolio/processor/FileProcessor.java @@ -1,14 +1,13 @@ package org.am.mypotrfolio.processor; +import org.am.mypotrfolio.domain.common.DocumentRequest; import org.springframework.web.multipart.MultipartFile; -import com.am.common.amcommondata.model.enums.BrokerType; - import java.util.List; import java.util.Map; public interface FileProcessor { - List> processFile(MultipartFile file, BrokerType brokerType); + List> processFile(MultipartFile file, DocumentRequest documentRequest); String getFileType(); boolean canProcess(String fileExtension); } diff --git a/src/main/java/org/am/mypotrfolio/service/DocumentProcessorService.java b/src/main/java/org/am/mypotrfolio/service/DocumentProcessorService.java index 0cbc33b..f787cd9 100644 --- a/src/main/java/org/am/mypotrfolio/service/DocumentProcessorService.java +++ b/src/main/java/org/am/mypotrfolio/service/DocumentProcessorService.java @@ -25,8 +25,8 @@ public class DocumentProcessorService { private final Map processStatusMap = new ConcurrentHashMap<>(); private final DocumentProcessor documentProcessor; - public DocumentProcessResponse processDocument(MultipartFile file, DocumentType documentType) { - var documentRequest = getDocumentRequest(file, documentType); + public DocumentProcessResponse processDocument(MultipartFile file, DocumentType documentType, String portfolioId, String userId) { + var documentRequest = getDocumentRequest(file, documentType, portfolioId, userId); log.info("[ProcessId: {}] Starting document processing for type: {}", documentRequest.getRequestId(), documentType); processStatusMap.put(documentRequest.getRequestId(), ProcessingStatus.QUEUED); @@ -37,7 +37,7 @@ public DocumentProcessResponse processDocument(MultipartFile file, DocumentType processStatusMap.put(documentRequest.getRequestId(), ProcessingStatus.PROCESSING); log.info("[ProcessId: {}] Processing document with {} processor", documentRequest.getRequestId(), documentRequest.getBrokerType()); - DocumentProcessResponse response = documentProcessor.processDocument(documentRequest); + DocumentProcessResponse response = documentProcessor.processDocument(documentRequest, portfolioId, userId); response.setProcessId(documentRequest.getRequestId()); response.setStatus(ProcessingStatus.COMPLETED); processStatusMap.put(documentRequest.getRequestId(), ProcessingStatus.COMPLETED); @@ -52,19 +52,19 @@ public DocumentProcessResponse processDocument(MultipartFile file, DocumentType } } - private DocumentRequest getDocumentRequest(MultipartFile file, DocumentType documentType) { + private DocumentRequest getDocumentRequest(MultipartFile file, DocumentType documentType, String portfolioId, String userId) { UUID processId = UUID.randomUUID(); BrokerType brokerType = detectBrokerType(file); - return DocumentRequest.builder().file(file).documentType(documentType).requestId(processId).brokerType(brokerType).build(); + return DocumentRequest.builder().file(file).documentType(documentType).requestId(processId).brokerType(brokerType).portfolioId(portfolioId).userId(userId).build(); } - public List processBatchDocuments(List files, DocumentType documentType) { + public List processBatchDocuments(List files, DocumentType documentType, String portfolioId, String userId) { UUID batchId = UUID.randomUUID(); log.info("[BatchId: {}] Starting batch processing of {} documents", batchId, files.size()); List responses = new ArrayList<>(); for (MultipartFile file : files) { - responses.add(processDocument(file, documentType)); + responses.add(processDocument(file, documentType, portfolioId, userId)); } log.info("[BatchId: {}] Completed batch processing", batchId); diff --git a/src/main/java/org/am/mypotrfolio/service/MessagingEventService.java b/src/main/java/org/am/mypotrfolio/service/MessagingEventService.java index a4add7a..b18d4e5 100644 --- a/src/main/java/org/am/mypotrfolio/service/MessagingEventService.java +++ b/src/main/java/org/am/mypotrfolio/service/MessagingEventService.java @@ -5,7 +5,11 @@ import java.util.UUID; import org.am.mypotrfolio.kafka.model.PortfolioUpdateEvent; +import org.am.mypotrfolio.kafka.model.TradeUpdateEvent; import org.am.mypotrfolio.kafka.producer.KafkaProducerService; +import org.am.mypotrfolio.model.trade.FNOTradeType; +import org.am.mypotrfolio.model.trade.TradeModel; +import org.am.mypotrfolio.model.trade.TradeType; import org.am.mypotrfolio.utils.ObjectUtils; import org.springframework.stereotype.Service; @@ -22,29 +26,68 @@ public class MessagingEventService { private final KafkaProducerService kafkaProducerService; - public void sendStockPortfolioMessage(List assetModels, UUID processId, BrokerType brokerType) { - var portfolioUpdateEvent = buildPortfolioUpdateEvent(processId, brokerType); + public void sendStockPortfolioMessage(List assetModels, UUID processId, BrokerType brokerType, String portfolioId, String userId) { + var portfolioUpdateEvent = buildPortfolioUpdateEvent(processId, brokerType, portfolioId, userId); portfolioUpdateEvent.setEquities(assetModels); kafkaProducerService.sendMessage(portfolioUpdateEvent); } - public void sendMutualFundPortfolioMessage(List mFundModels, UUID processId, BrokerType brokerType) { - var portfolioUpdateEvent = buildPortfolioUpdateEvent(processId, brokerType); + public void sendMutualFundPortfolioMessage(List mFundModels, UUID processId, BrokerType brokerType, String portfolioId, String userId) { + var portfolioUpdateEvent = buildPortfolioUpdateEvent(processId, brokerType, portfolioId, userId); portfolioUpdateEvent.setMutualFunds(mFundModels); kafkaProducerService.sendMessage(portfolioUpdateEvent); } + public void sendTradeFnoMessage(List trades, UUID processId, BrokerType brokerType, String portfolioId, String userId) { + var tradeUpdateEvent = buildTradeUpdateEvent(processId, brokerType, portfolioId, userId); + tradeUpdateEvent.setTrades(trades); + kafkaProducerService.sendTradeUpdateEvent(tradeUpdateEvent); + log.info("[ProcessId: {}] Successfully sent F&O trade update event with {} trades", processId, trades.size()); + } + + private FNOTradeType extractTradeType(List trades) { + return trades.stream().findFirst().map(trade -> trade.getInstrumentInfo().getSegment()).map(segment -> { + if(segment.equals("F")) { + return FNOTradeType.FUTIDX; + } else if(segment.equals("O")) { + return FNOTradeType.OPTIDX; + } else if(segment.equals("E")) { + return FNOTradeType.FUTEQ; + } else { + return FNOTradeType.OPTEQ; + } + }).orElse(null); + } + + public void sendTradeEqMessage(List trades, UUID processId, BrokerType brokerType, String portfolioId, String userId) { + var tradeUpdateEvent = buildTradeUpdateEvent(processId, brokerType, portfolioId, userId); + tradeUpdateEvent.setTrades(trades); + kafkaProducerService.sendTradeUpdateEvent(tradeUpdateEvent); + log.info("[ProcessId: {}] Successfully sent equity trade update event with {} trades", processId, trades.size()); + } + public void sendMessage(PortfolioUpdateEvent portfolioUpdateEvent, UUID processId, BrokerType brokerType) { log.info("[ProcessId: {}] Preparing to send portfolio update event and payload {}", processId, ObjectUtils.convertToJson(portfolioUpdateEvent)); kafkaProducerService.sendMessage(portfolioUpdateEvent); log.info("[ProcessId: {}] Successfully sent portfolio update event", processId); } - private PortfolioUpdateEvent buildPortfolioUpdateEvent(UUID processId, BrokerType brokerType) { + private PortfolioUpdateEvent buildPortfolioUpdateEvent(UUID processId, BrokerType brokerType, String portfolioId, String userId) { return PortfolioUpdateEvent.builder() .id(processId) - .userId("MKU257") + .userId(userId) + .brokerType(brokerType) + .timestamp(LocalDateTime.now()) + .portfolioId(portfolioId) + .build(); + } + + private TradeUpdateEvent buildTradeUpdateEvent(UUID processId, BrokerType brokerType, String portfolioId, String userId) { + return TradeUpdateEvent.builder() + .id(processId) + .userId(userId) .brokerType(brokerType) + .portfolioId(portfolioId) .timestamp(LocalDateTime.now()) .build(); } diff --git a/src/main/java/org/am/mypotrfolio/service/NseService.java b/src/main/java/org/am/mypotrfolio/service/NseService.java index 9ad0930..35bb527 100644 --- a/src/main/java/org/am/mypotrfolio/service/NseService.java +++ b/src/main/java/org/am/mypotrfolio/service/NseService.java @@ -7,7 +7,7 @@ import org.am.mypotrfolio.domain.common.NseSecurity; import org.am.mypotrfolio.domain.common.DocumentRequest; -import org.am.mypotrfolio.nsesecurity.repo.EquityDataRepository; + import org.am.mypotrfolio.processor.FileProcessorFactory; import org.am.mypotrfolio.service.mapper.NseSecurityMapper; import org.springframework.stereotype.Service; @@ -27,7 +27,7 @@ public class NseService { private final FileProcessorFactory fileProcessorFactory; private final NseSecurityMapper securityMapper; - private final EquityDataRepository equityDataRepository; + private final SecurityService securityService; private final ObjectMapper objectMapper; @@ -68,14 +68,17 @@ public List processNseSecurityFile(List> file // Try to find NSE security by name or other identifiers var securityModel = securityMapper.toSecurityModel(security); - equityDataRepository.findByKey(security.getIsin()) - .ifPresentOrElse( - equityData -> { - securityModel.getMetadata().setMarketCapValue(equityData.getMarketCap()); - log.debug("[ProcessId: {}] Found market cap data for security: {}", processId, security.getSecurityId()); - }, - () -> log.debug("[ProcessId: {}] No market cap data found for security: {}", processId, security.getSecurityId()) - ); + + //@TODO: Find api/doc to provide realtime market data for all securities + + // equityDataRepository.findByKey(security.getIsin()) + // .ifPresentOrElse( + // equityData -> { + // securityModel.getMetadata().setMarketCapValue(equityData.getMarketCap()); + // log.debug("[ProcessId: {}] Found market cap data for security: {}", processId, security.getSecurityId()); + // }, + // () -> log.debug("[ProcessId: {}] No market cap data found for security: {}", processId, security.getSecurityId()) + // ); securityModels.add(securityModel); } diff --git a/src/main/java/org/am/mypotrfolio/service/TradeService.java b/src/main/java/org/am/mypotrfolio/service/TradeService.java new file mode 100644 index 0000000..79ec35b --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/service/TradeService.java @@ -0,0 +1,11 @@ +package org.am.mypotrfolio.service; + +import java.util.List; + +import org.am.mypotrfolio.domain.common.DocumentRequest; +import org.am.mypotrfolio.model.trade.TradeModel; + +public interface TradeService { + + List processTradeFile(DocumentRequest documentRequest); +} diff --git a/src/main/java/org/am/mypotrfolio/service/PortfolioServiceImpl.java b/src/main/java/org/am/mypotrfolio/service/impl/PortfolioServiceImpl.java similarity index 91% rename from src/main/java/org/am/mypotrfolio/service/PortfolioServiceImpl.java rename to src/main/java/org/am/mypotrfolio/service/impl/PortfolioServiceImpl.java index 3e599de..4dbd798 100644 --- a/src/main/java/org/am/mypotrfolio/service/PortfolioServiceImpl.java +++ b/src/main/java/org/am/mypotrfolio/service/impl/PortfolioServiceImpl.java @@ -1,19 +1,16 @@ -package org.am.mypotrfolio.service; +package org.am.mypotrfolio.service.impl; import java.util.ArrayList; -import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Optional; -import java.util.Set; import java.util.UUID; +import java.util.Optional; import org.am.mypotrfolio.domain.common.MutualFundAsset; import org.am.mypotrfolio.domain.common.DocumentRequest; import org.am.mypotrfolio.domain.common.StockAsset; -import org.am.mypotrfolio.nsesecurity.domain.NseSecurity; -import org.am.mypotrfolio.nsesecurity.repo.NseSecurityRepository; import org.am.mypotrfolio.processor.FileProcessorFactory; +import org.am.mypotrfolio.service.PortfolioService; import org.springframework.stereotype.Service; import com.am.common.amcommondata.model.asset.equity.EquityModel; @@ -29,13 +26,12 @@ import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; -@Service +@Service("documentProcessorPortfolioService") @Slf4j @RequiredArgsConstructor public class PortfolioServiceImpl implements PortfolioService { private final FileProcessorFactory fileProcessorFactory; private final SecurityService securityService; - private final NseSecurityRepository nseSecurityRepository; private final ObjectMapper objectMapper; @@ -47,7 +43,7 @@ public List processEquityFile(DocumentRequest portfolioRequest) { // Process the file using appropriate processor log.debug("[ProcessId: {}] Getting file processor for file type", portfolioRequest.getRequestId()); List> fileData = fileProcessorFactory.getProcessor(portfolioRequest.getFile()) - .processFile(portfolioRequest.getFile(), portfolioRequest.getBrokerType()); + .processFile(portfolioRequest.getFile(), portfolioRequest); log.debug("[ProcessId: {}] Successfully processed file data, converting to StockPortfolio objects", portfolioRequest.getRequestId()); return processPortfolioFileAndGetAssets(fileData, portfolioRequest.getBrokerType(), portfolioRequest.getRequestId()); } catch (Exception e) { @@ -63,7 +59,7 @@ public List processMutualFundFile(DocumentRequest portfolioRequ // Process the file using appropriate processor log.debug("[ProcessId: {}] Getting file processor for file type", portfolioRequest.getRequestId()); List> fileData = fileProcessorFactory.getProcessor(portfolioRequest.getFile()) - .processFile(portfolioRequest.getFile(), portfolioRequest.getBrokerType()); + .processFile(portfolioRequest.getFile(), portfolioRequest); log.debug("[ProcessId: {}] Successfully processed file data, converting to StockPortfolio objects", portfolioRequest.getRequestId()); return processMutualFundsPortfolioFileAndGetAssets(fileData, portfolioRequest.getBrokerType(), portfolioRequest.getRequestId()); } catch (Exception e) { @@ -71,7 +67,7 @@ public List processMutualFundFile(DocumentRequest portfolioRequ throw e; } } - + @SneakyThrows public List processMutualFundsPortfolioFileAndGetAssets(List> fileData, BrokerType brokerType, UUID processId) { // Convert the data to StockPortfolio objects @@ -139,10 +135,10 @@ private EquityModel getAssetModel(StockAsset stock, BrokerType brokerType) { .name(stock.getName()); if(brokerType.isDhan() || brokerType.isMStock()){ - Optional nseSecurity = nseSecurityRepository.findBestMatchBySearchParam(brokerType.isDhan() ? stock.getName() : stock.getSymbol()); + Optional nseSecurity = findBestMatchBySearchParam(brokerType.isDhan() ? stock.getName() : stock.getSymbol()); if (nseSecurity.isPresent()) { - NseSecurity security = nseSecurity.get(); - stock.setIsin(security.getIsin()); + SecurityModel security = nseSecurity.get(); + stock.setIsin(security.getKey().getIsin()); } } @@ -161,4 +157,12 @@ private EquityModel getAssetModel(StockAsset stock, BrokerType brokerType) { //} return assetBuilder.build(); } + + Optional findBestMatchBySearchParam(String searchParam) { + if (searchParam == null || searchParam.trim().isEmpty()) { + return Optional.empty(); + } + List matches = securityService.findSecurityBySearchParam(searchParam); + return matches.isEmpty() ? Optional.empty() : Optional.of(matches.get(0)); + } } diff --git a/src/main/java/org/am/mypotrfolio/service/impl/TradeServiceImpl.java b/src/main/java/org/am/mypotrfolio/service/impl/TradeServiceImpl.java new file mode 100644 index 0000000..2e2c1f6 --- /dev/null +++ b/src/main/java/org/am/mypotrfolio/service/impl/TradeServiceImpl.java @@ -0,0 +1,181 @@ +package org.am.mypotrfolio.service.impl; + +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.stream.Collectors; + +import org.am.mypotrfolio.domain.common.DocumentRequest; +import org.am.mypotrfolio.mapper.TradeMapper; +import org.am.mypotrfolio.model.trade.Trade; +import org.am.mypotrfolio.model.trade.TradeModel; +import org.am.mypotrfolio.processor.FileProcessorFactory; +import org.am.mypotrfolio.service.TradeService; +import org.springframework.stereotype.Service; + +import com.am.common.amcommondata.model.enums.BrokerType; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; + +import lombok.RequiredArgsConstructor; +import lombok.SneakyThrows; +import lombok.extern.slf4j.Slf4j; + +/** + * Implementation of the TradeService for processing trade files + * Based on Zerodha's F&O trade book structure + */ +@Service +@RequiredArgsConstructor +@Slf4j +public class TradeServiceImpl implements TradeService { + + private final FileProcessorFactory fileProcessorFactory; + private final ObjectMapper objectMapper; + + @Override + public List processTradeFile(DocumentRequest documentRequest) { + + log.info("[ProcessId: {}] Starting to process trade file: {}", documentRequest.getRequestId(), documentRequest.getFile().getOriginalFilename()); + try { + // Process the file using appropriate processor + log.debug("[ProcessId: {}] Getting file processor for file type", documentRequest.getRequestId()); + List> fileData = fileProcessorFactory.getProcessor(documentRequest.getFile()) + .processFile(documentRequest.getFile(), documentRequest); + log.debug("[ProcessId: {}] Successfully processed file data, converting to StockPortfolio objects", documentRequest.getRequestId()); + return processTradeFileAndGetTradeList(fileData, documentRequest.getBrokerType(), documentRequest.getRequestId()); + } catch (Exception e) { + log.error("[ProcessId: {}] Error processing trade file: {}", documentRequest.getRequestId(), e.getMessage(), e); + throw e; + } + } + + @SneakyThrows + public List processTradeFileAndGetTradeList(List> fileData, BrokerType brokerType, UUID processId) { + TradeMapper tradeMapper = new TradeMapper(); + // Convert the data to StockPortfolio objects + String payload = objectMapper.writeValueAsString(fileData); + List trades = objectMapper.readValue(payload, new TypeReference>() {}); + + // For Zerodha, aggregate trades with the same order ID + if (BrokerType.ZERODHA.equals(brokerType)) { + trades = aggregateZerodhaTrades(trades, processId); + } + + // Convert to TradeModels + List tradeList = new ArrayList<>(); + for (Trade trade : trades) { + log.debug("[ProcessId: {}] Processing trade: {}", processId, trade.getOrderId()); + TradeModel tradeModel = tradeMapper.toTradeModel(trade, brokerType); + tradeList.add(tradeModel); + } + log.info("[ProcessId: {}] Successfully processed {} trade entries", processId, tradeList.size()); + return tradeList; + } + + /** + * Aggregates Zerodha trades with the same order ID. + * For trades with the same order ID: + * - Quantities are summed + * - Prices are weighted averaged based on quantity + * - Other attributes are taken from the first trade + * + * @param trades List of trades to aggregate + * @param processId Process ID for logging + * @return List of aggregated trades + */ + private List aggregateZerodhaTrades(List trades, UUID processId) { + if (trades == null || trades.isEmpty()) { + return trades; + } + + log.debug("[ProcessId: {}] Aggregating Zerodha trades by order ID", processId); + + // Group trades by order ID + Map> tradesByOrderId = trades.stream() + .filter(trade -> trade.getOrderId() != null && !trade.getOrderId().isEmpty()) + .collect(Collectors.groupingBy(Trade::getOrderId)); + + List aggregatedTrades = new ArrayList<>(); + + // Process each group of trades with the same order ID + for (Map.Entry> entry : tradesByOrderId.entrySet()) { + String orderId = entry.getKey(); + List tradesWithSameOrderId = entry.getValue(); + + if (tradesWithSameOrderId.size() == 1) { + // If there's only one trade with this order ID, no need to aggregate + aggregatedTrades.add(tradesWithSameOrderId.get(0)); + } else { + // Aggregate trades with the same order ID + Trade aggregatedTrade = aggregateTradesWithSameOrderId(tradesWithSameOrderId); + aggregatedTrades.add(aggregatedTrade); + log.debug("[ProcessId: {}] Aggregated {} trades with order ID: {}", + processId, tradesWithSameOrderId.size(), orderId); + } + } + + // Add any trades without order IDs + trades.stream() + .filter(trade -> trade.getOrderId() == null || trade.getOrderId().isEmpty()) + .forEach(aggregatedTrades::add); + + log.info("[ProcessId: {}] Aggregated {} trades into {} trades", + processId, trades.size(), aggregatedTrades.size()); + + return aggregatedTrades; + } + + /** + * Aggregates a list of trades with the same order ID into a single trade. + * + * @param trades List of trades with the same order ID + * @return A single aggregated trade + */ + private Trade aggregateTradesWithSameOrderId(List trades) { + if (trades == null || trades.isEmpty()) { + return null; + } + + // Use the first trade as the base + Trade baseTrade = trades.get(0); + + // Calculate total quantity and weighted average price + BigDecimal totalQuantity = BigDecimal.ZERO; + BigDecimal weightedPriceSum = BigDecimal.ZERO; + + for (Trade trade : trades) { + // Sum up quantities + totalQuantity = totalQuantity.add(trade.getQuantity()); + + // Calculate weighted price + BigDecimal weightedPrice = trade.getPrice().multiply(trade.getQuantity()); + weightedPriceSum = weightedPriceSum.add(weightedPrice); + } + + // Calculate weighted average price + BigDecimal averagePrice = totalQuantity.compareTo(BigDecimal.ZERO) > 0 ? + weightedPriceSum.divide(totalQuantity, 2, RoundingMode.HALF_UP) : BigDecimal.ZERO; + + // Create a new aggregated trade + return Trade.builder() + .symbol(baseTrade.getSymbol()) + .isin(baseTrade.getIsin()) + .tradeDate(baseTrade.getTradeDate()) + .exchange(baseTrade.getExchange()) + .segment(baseTrade.getSegment()) + .series(baseTrade.getSeries()) + .tradeType(baseTrade.getTradeType()) + .auction(baseTrade.getAuction()) + .quantity(totalQuantity) + .price(averagePrice) + .tradeId(baseTrade.getTradeId()) + .orderId(baseTrade.getOrderId()) + .orderExecutionTime(baseTrade.getOrderExecutionTime()) + .build(); + } +} diff --git a/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessor.java b/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessor.java index 34480fb..32820d0 100644 --- a/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessor.java +++ b/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessor.java @@ -6,6 +6,6 @@ import com.am.common.amcommondata.model.enums.BrokerType; public interface DocumentProcessor { - DocumentProcessResponse processDocument(DocumentRequest documentRequest); + DocumentProcessResponse processDocument(DocumentRequest documentRequest, String portfolioId, String userId); BrokerType getBrokerType(); } diff --git a/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessorImpl.java b/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessorImpl.java index 1d5781d..7ab057c 100644 --- a/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessorImpl.java +++ b/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessorImpl.java @@ -1,16 +1,16 @@ package org.am.mypotrfolio.service.processor; import java.util.List; -import java.util.Set; import org.am.mypotrfolio.domain.common.DocumentRequest; import org.am.mypotrfolio.model.DocumentProcessResponse; +import org.am.mypotrfolio.model.trade.TradeModel; import org.am.mypotrfolio.service.MessagingEventService; import org.am.mypotrfolio.service.NseService; import org.am.mypotrfolio.service.PortfolioService; +import org.am.mypotrfolio.service.TradeService; import org.springframework.stereotype.Component; -import com.am.common.amcommondata.model.asset.AssetModel; import com.am.common.amcommondata.model.asset.equity.EquityModel; import com.am.common.amcommondata.model.asset.mutualfund.MutualFundModel; import com.am.common.amcommondata.model.enums.BrokerType; @@ -21,13 +21,15 @@ @Component @RequiredArgsConstructor public class DocumentProcessorImpl implements DocumentProcessor { + @org.springframework.beans.factory.annotation.Qualifier("documentProcessorPortfolioService") private final PortfolioService portfolioService; private final NseService nseService; private final MessagingEventService messagingEventService; + private final TradeService tradeService; @Override - public DocumentProcessResponse processDocument(DocumentRequest documentRequest) { - processPortfolio(documentRequest); + public DocumentProcessResponse processDocument(DocumentRequest documentRequest, String portfolioId, String userId) { + processPortfolio(documentRequest, portfolioId, userId); DocumentProcessResponse response = new DocumentProcessResponse(); response.setDocumentType(documentRequest.getDocumentType().name()); response.setFileName(documentRequest.getFile().getOriginalFilename()); @@ -35,30 +37,43 @@ public DocumentProcessResponse processDocument(DocumentRequest documentRequest) return response; } - private void processPortfolio(DocumentRequest documentRequest) { + private void processPortfolio(DocumentRequest documentRequest, String portfolioId, String userId) { if(documentRequest.getDocumentType().isStockPortfolio()) { - processEquityPortfolio(documentRequest); + processEquityPortfolio(documentRequest, portfolioId, userId); } else if(documentRequest.getDocumentType().isMutualFund()) { - processMutualFundsPortfolio(documentRequest); - } - else if(documentRequest.getDocumentType().isNseIndices()) { - processNseIndices(documentRequest); + processMutualFundsPortfolio(documentRequest, portfolioId, userId); + } else if(documentRequest.getDocumentType().isNseIndices()) { + processNseIndices(documentRequest, portfolioId); + } else if(documentRequest.getDocumentType().isTradeFno()) { + processTradeFno(documentRequest, portfolioId, userId); + } else if(documentRequest.getDocumentType().isTradeEq()) { + processTradeEq(documentRequest, portfolioId, userId); } } - private void processEquityPortfolio(DocumentRequest documentRequest) { + private void processEquityPortfolio(DocumentRequest documentRequest, String portfolioId, String userId) { List assets = portfolioService.processEquityFile(documentRequest); - messagingEventService.sendStockPortfolioMessage(assets, documentRequest.getRequestId(), documentRequest.getBrokerType()); + messagingEventService.sendStockPortfolioMessage(assets, documentRequest.getRequestId(), documentRequest.getBrokerType(), portfolioId, userId); } - private void processNseIndices(DocumentRequest documentRequest) { + private void processNseIndices(DocumentRequest documentRequest, String portfolioId) { List assets = nseService.processNseSecurity(documentRequest); //messagingEventService.sendNseIndicesMessage(assets, documentRequest.getRequestId(), documentRequest.getBrokerType()); } - private void processMutualFundsPortfolio(DocumentRequest documentRequest) { + private void processMutualFundsPortfolio(DocumentRequest documentRequest, String portfolioId, String userId) { List mutualFunds = portfolioService.processMutualFundFile(documentRequest); - messagingEventService.sendMutualFundPortfolioMessage(mutualFunds, documentRequest.getRequestId(), documentRequest.getBrokerType()); + messagingEventService.sendMutualFundPortfolioMessage(mutualFunds, documentRequest.getRequestId(), documentRequest.getBrokerType(), portfolioId, userId); + } + + private void processTradeFno(DocumentRequest documentRequest, String portfolioId, String userId) { + List trades = tradeService.processTradeFile(documentRequest); + messagingEventService.sendTradeFnoMessage(trades, documentRequest.getRequestId(), documentRequest.getBrokerType(), portfolioId, userId); + } + + private void processTradeEq(DocumentRequest documentRequest, String portfolioId, String userId) { + List trades = tradeService.processTradeFile(documentRequest); + messagingEventService.sendTradeEqMessage(trades, documentRequest.getRequestId(), documentRequest.getBrokerType(), portfolioId, userId); } @Override diff --git a/src/main/resources/application-lot-sizes.yml b/src/main/resources/application-lot-sizes.yml new file mode 100644 index 0000000..0429f40 --- /dev/null +++ b/src/main/resources/application-lot-sizes.yml @@ -0,0 +1,51 @@ +trade: + lot-sizes: + # Default lot sizes for when no specific date-based configuration is found + default-lot-sizes: + NIFTY: 75 + BANKNIFTY: 30 + FINNIFTY: 65 + MIDCPNIFTY: 120 + + # Historical lot sizes by index with effective dates + indices: + NIFTY: + "2000-06-01": 200 + "2005-04-01": 100 + "2007-02-23": 50 + "2010-04-29": 25 + "2015-08-28": 30 + "2016-04-29": 40 + "2018-10-26": 20 + "2020-05-04": 25 + "2024-10-18": 75 + BANKNIFTY: + "2000-06-01": 200 + "2005-04-01": 100 + "2007-02-23": 50 + "2010-04-29": 25 + "2015-08-28": 25 + "2016-04-29": 25 + "2018-10-26": 25 + "2023-06-01": 15 + "2024-10-18": 30 + FINNIFTY: + "2000-06-01": 200 + "2005-04-01": 100 + "2007-02-23": 40 + "2010-04-29": 40 + "2015-08-28": 40 + "2016-04-29": 40 + "2018-10-26": 40 + "2020-05-04": 40 + "2024-10-18": 25 + MIDCPNIFTY: + "2000-06-01": 200 + "2005-04-01": 100 + "2007-02-23": 75 + "2010-04-29": 75 + "2015-08-28": 75 + "2016-04-29": 75 + "2018-10-26": 75 + "2020-05-04": 75 + "2024-10-18": 50 diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index ae43139..f00a083 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -1,41 +1,44 @@ +MONGODB_URL: mongodb://ssd2658:ssd2658@localhost:27017/portfolio?authSource=admin + +# Kafka Configuration +KAFKA_BOOTSTRAP_SERVERS: localhost:9092 +KAFKA_SECURITY_PROTOCOL: PLAINTEXT +KAFKA_SASL_MECHANISM: PLAIN +KAFKA_SASL_JAAS_CONFIG: #org.apache.kafka.common.security.plain.PlainLoginModule required username="kafkaUser" password="kafkaPassword123!"; + server: port: 8080 -spring: +spring: servlet: multipart: max-file-size: 10MB max-request-size: 10MB application: name: document-processor-data - datasource: - url: jdbc:postgresql://localhost:5432/postgres - username: postgres - password: password - driver-class-name: org.postgresql.Driver - jpa: - database-platform: org.hibernate.dialect.PostgreSQLDialect - hibernate: - ddl-auto: update - show-sql: true - properties: - hibernate: - format_sql: true - kafka: - bootstrap-servers: localhost:9092 - consumer: - group-id: document-processor - auto-offset-reset: earliest + profiles: + include: + - lot-sizes data: mongodb: + uri: ${MONGODB_URL} + database: portfolio auto-index-creation: true + kafka: + bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS} + properties: + security.protocol: ${KAFKA_SECURITY_PROTOCOL} + sasl.mechanism: ${KAFKA_SASL_MECHANISM} + sasl.jaas.config: ${KAFKA_SASL_JAAS_CONFIG} + consumer: + group-id: am-document-processor-group + auto-offset-reset: earliest + listener: + missing-topics-fatal: false app: kafka: - topic: am-portfolio-topic - persistence: - mongodb: - uri: mongodb://admin:password@localhost:27017/nse_security?authSource=admin - database: nse_security + portfolio-topic: am-portfolio + trade-topic: am-trade logging: level: @@ -45,4 +48,35 @@ logging: org.hibernate: INFO org.hibernate.SQL: INFO '[org.hibernate.type.descriptor.sql.BasicBinder]': TRACE - '[org.springframework.data.mongodb.core.MongoTemplate]': DEBUG \ No newline at end of file + '[org.springframework.data.mongodb.core.MongoTemplate]': DEBUG + + + +# Management/Actuator Configuration +management: + endpoints: + web: + exposure: + include: health,info,metrics,prometheus + endpoint: + health: + probes: + enabled: true + show-details: always + validate-group-membership: false + group: + readiness: + include: diskSpace,mongodb,redis,kafka + liveness: + include: diskSpace,mongodb,redis,kafka + health: + livenessstate: + enabled: true + readinessstate: + enabled: true + mongo: + enabled: true + redis: + enabled: true + kafka: + enabled: true \ No newline at end of file